GStreamer多媒体框架深入解析与实战应用

GStreamer是一个功能强大的多媒体框架，基于管道（Pipeline）架构设计，支持音视频的采集、处理、编解码、传输和播放。本文将深入探讨GStreamer的核心架构、关键组件和实际应用，并通过Python代码示例展示其强大的功能。

1. GStreamer架构概述

1.1 核心设计理念

GStreamer采用基于图（Graph）的管道架构，将多媒体处理分解为一系列独立的元素（Elements），通过连接这些元素形成处理管道：

# GStreamer Python绑定导入和初始化详解

# 导入GObject Introspection库，用于访问C库的Python绑定
import gi

# 指定GStreamer版本，确保使用正确的API版本
# GStreamer 1.0是当前主要版本，与0.10版本不兼容
gi.require_version('Gst', '1.0')

# 导入GStreamer核心模块
from gi.repository import Gst, GObject, GLib

# 导入Python标准库和第三方库
import threading    # 多线程支持
import time        # 时间处理
from typing import Dict, List, Optional, Callable, Any  # 类型注解
from enum import Enum        # 枚举类型
from dataclasses import dataclass  # 数据类
import json        # JSON处理
import numpy as np # 数值计算（用于音视频数据处理）

# 初始化GStreamer框架
# 这是使用GStreamer的第一步，必须在创建任何GStreamer对象之前调用
# None参数表示不传递命令行参数
Gst.init(None)

# 注意：Gst.init()会：
# 1. 初始化GStreamer核心库
# 2. 注册内置插件
# 3. 设置调试系统
# 4. 初始化类型系统

class ElementType(Enum):
    """
    GStreamer元素类型枚举
    
    GStreamer中的元素按功能分为不同类型，每种类型在管道中承担特定角色：
    - 源元素产生数据
    - 接收元素消费数据
    - 过滤元素处理数据
    - 编解码元素转换数据格式
    """
    SOURCE = "source"          # 数据源：产生音视频数据（如文件读取、摄像头采集）
    SINK = "sink"              # 数据接收器：消费音视频数据（如文件写入、屏幕显示）
    FILTER = "filter"          # 过滤器：处理数据但不改变格式（如音量调节、视频裁剪）
    CODEC = "codec"            # 编解码器：压缩或解压缩数据（如H.264编码、MP3解码）
    DEMUXER = "demuxer"        # 解复用器：分离容器中的音视频流（如MP4解复用）
    MUXER = "muxer"            # 复用器：将音视频流合并到容器（如MP4复用）
    CONVERTER = "converter"    # 格式转换器：改变数据格式（如YUV转RGB、采样率转换）
    EFFECT = "effect"          # 特效处理：添加视觉或音频效果（如模糊、回声）

class PadDirection(Enum):
    """
    Pad方向枚举
    
    Pad是GStreamer元素的输入输出接口：
    - 源Pad输出数据到下游元素
    - 接收Pad从上游元素接收数据
    - 元素通过Pad连接形成数据流管道
    """
    SRC = "src"    # 源Pad：元素的输出接口，向下游发送数据
    SINK = "sink"  # 接收Pad：元素的输入接口，从上游接收数据

@dataclass
class ElementInfo:
    """
    元素信息数据类
    
    封装GStreamer元素的完整信息，用于：
    - 元素注册和管理
    - 动态管道构建
    - 调试和监控
    - 配置验证
    """
    name: str                           # 元素实例名称（在管道中唯一）
    element_type: ElementType           # 元素功能类型
    factory_name: str                   # GStreamer工厂名称（用于创建元素）
    description: str                    # 元素功能描述
    src_pads: List[str]                # 源Pad列表（输出接口）
    sink_pads: List[str]               # 接收Pad列表（输入接口）
    properties: Dict[str, Any]         # 元素属性字典（配置参数）
    
    def __post_init__(self):
        """数据类初始化后处理"""
        # 验证必要字段
        if not self.name or not self.factory_name:
            raise ValueError("Element name and factory_name are required")
        
        # 确保Pad列表不为None
        if self.src_pads is None:
            self.src_pads = []
        if self.sink_pads is None:
            self.sink_pads = []
        
        # 确保属性字典不为None
        if self.properties is None:
            self.properties = {}

class GStreamerCore:
    """
    GStreamer核心管理器
    
    这个类封装了GStreamer管道的完整生命周期管理，包括：
    - 管道创建和销毁
    - 元素添加和连接
    - 状态管理和控制
    - 消息处理和事件回调
    - 错误处理和恢复
    
    设计模式：
    - 单例模式：确保管道的唯一性
    - 观察者模式：通过回调处理异步事件
    - 工厂模式：动态创建和配置元素
    """
    
    def __init__(self):
        """
        初始化GStreamer核心管理器
        
        设置管道基础设施和事件处理机制
        """
        # 核心组件初始化
        self.pipeline = None        # GStreamer管道对象（数据流的容器）
        self.bus = None            # 消息总线（用于异步通信）
        self.loop = None           # GLib主事件循环（处理异步消息）
        self.elements = {}         # 元素字典（name -> GstElement映射）
        self.is_playing = False    # 播放状态标志
        
        # 消息处理回调映射
        # GStreamer使用消息总线进行异步通信，不同类型的消息需要不同的处理逻辑
        self.message_handlers = {
            Gst.MessageType.ERROR: self._handle_error,              # 错误消息：处理管道错误
            Gst.MessageType.EOS: self._handle_eos,                  # 流结束消息：处理播放完成
            Gst.MessageType.STATE_CHANGED: self._handle_state_changed,  # 状态变化：跟踪元素状态
            Gst.MessageType.BUFFERING: self._handle_buffering,      # 缓冲消息：处理网络流缓冲
            Gst.MessageType.STREAM_STATUS: self._handle_stream_status   # 流状态：监控流的健康状态
        }
        
        # 用户自定义回调函数
        # 允许应用程序注册自定义的事件处理逻辑
        self.state_change_callback = None  # 状态变化回调：通知应用程序状态改变
        self.error_callback = None         # 错误回调：通知应用程序发生错误
        self.eos_callback = None           # 流结束回调：通知应用程序播放完成
        
        # 性能监控和调试
        self.start_time = None             # 管道启动时间
        self.frame_count = 0               # 处理的帧数统计
        self.error_count = 0               # 错误计数
        
        print("GStreamerCore initialized successfully")
    
    def create_pipeline(self, name: str = "main-pipeline") -> bool:
        """
        创建GStreamer管道
        
        管道是GStreamer的核心概念，它是一个容器，用于：
        - 管理所有元素的生命周期
        - 协调元素之间的数据流
        - 处理状态变化和同步
        - 提供统一的控制接口
        
        Args:
            name: 管道名称，用于调试和日志记录
            
        Returns:
            bool: 创建成功返回True，失败返回False
            
        管道创建过程：
        1. 创建Pipeline对象
        2. 获取消息总线
        3. 设置消息监听
        4. 连接消息处理回调
        """
        try:
            # 步骤1: 创建Pipeline对象
            # Pipeline继承自Bin，Bin继承自Element
            # Pipeline是顶级容器，可以包含其他元素和Bin
            self.pipeline = Gst.Pipeline.new(name)
            if not self.pipeline:
                print(f"Failed to create pipeline: {name}")
                return False
            
            # 步骤2: 获取消息总线
            # 总线是GStreamer的异步通信机制
            # 所有元素都可以向总线发送消息，应用程序通过总线接收消息
            self.bus = self.pipeline.get_bus()
            
            # 步骤3: 启用信号监听
            # add_signal_watch()启动一个GSource来监听总线消息
            # 这使得消息可以在主线程的事件循环中处理
            self.bus.add_signal_watch()
            
            # 步骤4: 连接消息处理回调
            # 当总线接收到消息时，会调用_on_message方法
            # 这是异步消息处理的入口点
            self.bus.connect("message", self._on_message)
            
            # 记录创建时间用于性能监控
            self.start_time = time.time()
            
            print(f"Pipeline '{name}' created successfully")
            print(f"Pipeline state: {self.pipeline.get_state(0)[1]}")
            return True
        
        except Exception as e:
            print(f"Error creating pipeline: {e}")
            self.error_count += 1
            return False
    
    def _on_message(self, bus, message):
        """
        消息总线回调函数
        
        这是GStreamer异步通信的核心，处理来自管道中所有元素的消息
        消息类型包括错误、警告、信息、状态变化、流结束等
        
        Args:
            bus: 消息总线对象
            message: 接收到的消息对象
        """
        message_type = message.type
        
        # 根据消息类型调用相应的处理函数
        handler = self.message_handlers.get(message_type)
        if handler:
            handler(message)
        else:
            # 处理未注册的消息类型
            print(f"Unhandled message type: {message_type}")
            
        # 消息处理完成后的清理
        # 注意：不要在这里释放message，GStreamer会自动管理
    
    def add_element(self, factory_name: str, element_name: str) -> Optional[Gst.Element]:
        """
        添加元素到管道
        
        元素是GStreamer的基本处理单元，每个元素执行特定的功能：
        - 源元素：产生数据（filesrc, v4l2src, audiotestsrc等）
        - 过滤元素：处理数据（videoconvert, audioresample等）
        - 接收元素：消费数据（filesink, autovideosink等）
        
        Args:
            factory_name: GStreamer元素工厂名称（如"filesrc", "h264parse"）
            element_name: 元素实例名称（在管道中必须唯一）
            
        Returns:
            Optional[Gst.Element]: 成功返回元素对象，失败返回None
            
        元素添加过程：
        1. 通过工厂创建元素实例
        2. 将元素添加到管道容器
        3. 注册元素到内部字典
        4. 返回元素引用供后续配置
        """
        try:
            # 步骤1: 通过ElementFactory创建元素
            # ElementFactory是GStreamer的工厂模式实现
            # 它根据factory_name查找并创建相应的元素类型
            element = Gst.ElementFactory.make(factory_name, element_name)
            if not element:
                print(f"Failed to create element: {factory_name}")
                print(f"Available factories: {self._list_available_factories()}")
                return None
            
            # 步骤2: 将元素添加到管道
            # 管道作为容器管理元素的生命周期
            # 添加后元素的状态会与管道同步
            if not self.pipeline.add(element):
                print(f"Failed to add element to pipeline: {element_name}")
                return None
            
            # 步骤3: 注册元素到内部字典
            # 便于后续通过名称快速查找和操作元素
            self.elements[element_name] = element
            
            # 步骤4: 记录元素信息用于调试
            element_class = element.get_factory().get_name()
            print(f"Element '{element_name}' ({factory_name}) added to pipeline")
            print(f"Element class: {element_class}")
            
            # 获取元素的Pad信息
            src_pads = [pad.get_name() for pad in element.srcpads]
            sink_pads = [pad.get_name() for pad in element.sinkpads]
            print(f"Source pads: {src_pads}, Sink pads: {sink_pads}")
            
            return element
        
        except Exception as e:
            print(f"Error adding element {element_name}: {e}")
            self.error_count += 1
            return None
    
    def _list_available_factories(self) -> List[str]:
        """
        列出可用的元素工厂
        
        用于调试和错误诊断，当元素创建失败时显示可用选项
        """
        registry = Gst.Registry.get()
        factories = registry.get_feature_list(Gst.ElementFactory)
        return [factory.get_name() for factory in factories[:10]]  # 只显示前10个
    
    def link_elements(self, src_element: str, sink_element: str, 
                     src_pad: str = None, sink_pad: str = None) -> bool:
        """
        连接两个元素
        
        元素连接是GStreamer管道构建的核心操作：
        - 数据流向：从源元素的输出pad流向接收元素的输入pad
        - 能力协商：连接时会进行caps（capabilities）协商
        - 连接类型：支持自动连接和指定pad连接
        
        Args:
            src_element: 源元素名称
            sink_element: 接收元素名称
            src_pad: 源元素的输出pad名称（可选）
            sink_pad: 接收元素的输入pad名称（可选）
            
        Returns:
            bool: 连接成功返回True，失败返回False
            
        连接过程：
        1. 验证元素存在性
        2. 选择连接方式（自动或指定pad）
        3. 执行pad连接和能力协商
        4. 验证连接结果
        """
        try:
            # 步骤1: 获取源元素和接收元素
            # 从内部字典中查找已添加的元素实例
            src = self.elements.get(src_element)
            sink = self.elements.get(sink_element)
            
            if not src or not sink:
                print(f"Elements not found: {src_element} or {sink_element}")
                print(f"Available elements: {list(self.elements.keys())}")
                return False
            
            # 步骤2: 根据参数选择连接方式
            if src_pad and sink_pad:
                # 指定Pad连接模式
                # 用于复杂场景，如多路输出、特定格式连接等
                print(f"Attempting pad-specific link: {src_element}:{src_pad} -> {sink_element}:{sink_pad}")
                
                # 获取指定的静态pad对象
                # 静态pad在元素创建时就存在，与动态pad相对
                src_pad_obj = src.get_static_pad(src_pad)
                sink_pad_obj = sink.get_static_pad(sink_pad)
                
                if not src_pad_obj:
                    print(f"Source pad '{src_pad}' not found on {src_element}")
                    self._list_element_pads(src, "source")
                    return False
                    
                if not sink_pad_obj:
                    print(f"Sink pad '{sink_pad}' not found on {sink_element}")
                    self._list_element_pads(sink, "sink")
                    return False
                
                # 执行pad级别的连接
                # 这会触发能力协商过程
                result = src_pad_obj.link(sink_pad_obj)
            else:
                # 自动连接模式
                # GStreamer会自动选择兼容的pad进行连接
                # 适用于简单的一对一连接场景
                print(f"Attempting auto-link: {src_element} -> {sink_element}")
                result = src.link(sink)
            
            # 步骤3: 检查连接结果
            if result == Gst.PadLinkReturn.OK:
                print(f"Successfully linked {src_element} -> {sink_element}")
                self._log_connection_info(src, sink, src_pad, sink_pad)
                return True
            else:
                print(f"Failed to link {src_element} -> {sink_element}: {result}")
                self._diagnose_link_failure(result, src, sink, src_pad, sink_pad)
                return False
        
        except Exception as e:
            print(f"Error linking elements: {e}")
            self.error_count += 1
            return False
    
    def _log_connection_info(self, src: Gst.Element, sink: Gst.Element, 
                           src_pad_name: str = None, sink_pad_name: str = None):
        """
        记录连接信息用于调试
        
        显示连接后的元素状态和能力信息
        """
        try:
            src_pad = src.get_static_pad(src_pad_name or "src")
            sink_pad = sink.get_static_pad(sink_pad_name or "sink")
            
            if src_pad:
                src_caps = src_pad.get_current_caps()
                if src_caps:
                    print(f"Source caps: {src_caps.to_string()}")
            
            if sink_pad:
                sink_caps = sink_pad.get_current_caps()
                if sink_caps:
                    print(f"Sink caps: {sink_caps.to_string()}")
        except Exception as e:
            print(f"Error logging connection info: {e}")
    
    def _diagnose_link_failure(self, result: Gst.PadLinkReturn, src: Gst.Element, 
                              sink: Gst.Element, src_pad_name: str = None, 
                              sink_pad_name: str = None):
        """
        诊断连接失败的原因
        
        根据返回值分析失败原因并提供解决建议
        """
        failure_reasons = {
            Gst.PadLinkReturn.WRONG_HIERARCHY: "Elements not in same bin",
            Gst.PadLinkReturn.WAS_LINKED: "Pad already linked",
            Gst.PadLinkReturn.WRONG_DIRECTION: "Incompatible pad directions",
            Gst.PadLinkReturn.NOFORMAT: "No compatible format found",
            Gst.PadLinkReturn.NOSCHED: "No compatible scheduling",
            Gst.PadLinkReturn.REFUSED: "Link refused by element"
        }
        
        reason = failure_reasons.get(result, f"Unknown error: {result}")
        print(f"Link failure reason: {reason}")
        
        # 显示元素的pad信息
        print("Analyzing element pads:")
        self._list_element_pads(src, "source")
        self._list_element_pads(sink, "sink")
    
    def _list_element_pads(self, element: Gst.Element, element_type: str):
        """
        列出元素的所有pad
        
        用于调试和错误诊断
        """
        try:
            pads = element.pads
            print(f"{element_type} element '{element.get_name()}' pads:")
            for pad in pads:
                direction = "SRC" if pad.get_direction() == Gst.PadDirection.SRC else "SINK"
                caps = pad.get_current_caps()
                caps_str = caps.to_string() if caps else "No caps negotiated"
                print(f"  {pad.get_name()} ({direction}): {caps_str}")
        except Exception as e:
            print(f"Error listing pads: {e}")
    
    def set_element_property(self, element_name: str, property_name: str, value: Any) -> bool:
        """
        设置元素属性
        
        元素属性控制元素的行为和配置：
        - 输入属性：如文件路径、网络地址、设备名称等
        - 编码属性：如比特率、分辨率、帧率等
        - 输出属性：如文件格式、质量设置等
        - 行为属性：如缓冲大小、超时设置等
        
        Args:
            element_name: 元素名称
            property_name: 属性名称
            value: 属性值（类型必须与属性定义匹配）
            
        Returns:
            bool: 设置成功返回True，失败返回False
            
        属性设置过程：
        1. 验证元素存在性
        2. 检查属性有效性
        3. 执行属性设置
        4. 验证设置结果
        """
        try:
            # 步骤1: 获取目标元素
            element = self.elements.get(element_name)
            if not element:
                print(f"Element not found: {element_name}")
                print(f"Available elements: {list(self.elements.keys())}")
                return False
            
            # 步骤2: 验证属性存在性（可选，用于调试）
            if not self._validate_property(element, property_name):
                print(f"Warning: Property '{property_name}' may not exist on {element_name}")
                self._list_element_properties(element)
            
            # 步骤3: 设置属性值
            # GObject属性系统会自动进行类型检查和转换
            old_value = element.get_property(property_name) if self._validate_property(element, property_name) else None
            element.set_property(property_name, value)
            
            # 步骤4: 验证设置结果
            new_value = element.get_property(property_name)
            print(f"Set {element_name}.{property_name}: {old_value} -> {new_value}")
            
            # 记录属性变更用于调试
            if hasattr(self, 'property_changes'):
                self.property_changes.append({
                    'element': element_name,
                    'property': property_name,
                    'old_value': old_value,
                    'new_value': new_value,
                    'timestamp': time.time()
                })
            
            return True
        
        except Exception as e:
            print(f"Error setting property {element_name}.{property_name} = {value}: {e}")
            self.error_count += 1
            return False
    
    def _validate_property(self, element: Gst.Element, property_name: str) -> bool:
        """
        验证元素是否具有指定属性
        
        用于属性设置前的验证，避免运行时错误
        """
        try:
            # 获取元素的GObject类
            element_class = element.__class__
            # 检查属性是否存在
            return hasattr(element_class.props, property_name)
        except Exception:
            return False
    
    def _list_element_properties(self, element: Gst.Element):
        """
        列出元素的所有可用属性
        
        用于调试和错误诊断
        """
        try:
            # 获取元素的所有属性
            properties = [prop.name for prop in element.list_properties()]
            print(f"Available properties for {element.get_name()}: {properties[:10]}...")  # 只显示前10个
        except Exception as e:
            print(f"Error listing properties: {e}")
    
    def get_element_property(self, element_name: str, property_name: str) -> Any:
        """
        获取元素属性
        
        属性获取用于：
        - 状态监控：检查元素当前配置
        - 调试诊断：验证属性设置是否生效
        - 动态调整：基于当前值进行计算和调整
        - 信息收集：获取元素运行时信息
        
        Args:
            element_name: 元素名称
            property_name: 属性名称
            
        Returns:
            Any: 属性值，失败返回None
            
        属性获取过程：
        1. 验证元素存在性
        2. 检查属性有效性
        3. 获取属性值
        4. 返回结果
        """
        try:
            # 步骤1: 获取目标元素
            element = self.elements.get(element_name)
            if not element:
                print(f"Element not found: {element_name}")
                print(f"Available elements: {list(self.elements.keys())}")
                return None
            
            # 步骤2: 验证属性存在性
            if not self._validate_property(element, property_name):
                print(f"Warning: Property '{property_name}' may not exist on {element_name}")
                self._list_element_properties(element)
                # 仍然尝试获取，因为某些动态属性可能无法通过验证检测到
            
            # 步骤3: 获取属性值
            # GObject属性系统会返回适当类型的值
            value = element.get_property(property_name)
            
            # 步骤4: 记录属性访问（用于调试）
            print(f"Get {element_name}.{property_name} = {value} (type: {type(value).__name__})")
            
            # 记录属性访问用于性能分析
            if hasattr(self, 'property_accesses'):
                self.property_accesses.append({
                    'element': element_name,
                    'property': property_name,
                    'value': value,
                    'timestamp': time.time()
                })
            
            return value
        
        except Exception as e:
            print(f"Error getting property {element_name}.{property_name}: {e}")
            self.error_count += 1
            return None
    
    def start_pipeline(self) -> bool:
        """
        启动管道
        
        管道状态转换是GStreamer的核心机制：
        - NULL -> READY: 分配资源，准备处理
        - READY -> PAUSED: 预处理，准备播放但不开始
        - PAUSED -> PLAYING: 开始数据流处理
        
        状态转换过程：
        1. 验证管道存在性
        2. 请求状态转换到PLAYING
        3. 等待状态转换完成
        4. 更新内部状态标志
        
        Returns:
            bool: 启动成功返回True，失败返回False
        """
        try:
            # 步骤1: 验证管道存在性
            if not self.pipeline:
                print("No pipeline to start")
                print("Please create a pipeline first using create_pipeline()")
                return False
            
            # 记录启动前的状态
            current_state = self.pipeline.get_state(0)[1]  # 获取当前状态，不等待
            print(f"Current pipeline state: {current_state}")
            
            # 步骤2: 请求状态转换到PLAYING
            # set_state是异步操作，会返回状态转换的结果
            print("Requesting pipeline state change to PLAYING...")
            ret = self.pipeline.set_state(Gst.State.PLAYING)
            
            # 步骤3: 检查状态转换结果
            if ret == Gst.StateChangeReturn.FAILURE:
                print("Failed to start pipeline - state change failed")
                self._diagnose_pipeline_failure()
                return False
            elif ret == Gst.StateChangeReturn.ASYNC:
                print("Pipeline state change is asynchronous, waiting...")
                # 等待状态转换完成（最多等待5秒）
                ret = self.pipeline.get_state(5 * Gst.SECOND)[0]
                if ret == Gst.StateChangeReturn.FAILURE:
                    print("Pipeline failed to reach PLAYING state")
                    return False
            elif ret == Gst.StateChangeReturn.SUCCESS:
                print("Pipeline state changed to PLAYING immediately")
            
            # 步骤4: 更新内部状态
            self.is_playing = True
            self.start_time = time.time()
            
            # 验证最终状态
            final_state = self.pipeline.get_state(0)[1]
            print(f"Pipeline started successfully - final state: {final_state}")
            
            return True
        
        except Exception as e:
            print(f"Error starting pipeline: {e}")
            self.error_count += 1
            return False
    
    def _diagnose_pipeline_failure(self):
        """
        诊断管道启动失败的原因
        
        分析管道状态和元素状态，帮助定位问题
        """
        try:
            print("Diagnosing pipeline failure:")
            
            # 检查管道状态
            state_ret, current_state, pending_state = self.pipeline.get_state(0)
            print(f"Pipeline state: current={current_state}, pending={pending_state}")
            
            # 检查各个元素的状态
            print("Element states:")
            for name, element in self.elements.items():
                element_state = element.get_state(0)[1]
                print(f"  {name}: {element_state}")
            
            # 检查消息总线中的错误信息
            if hasattr(self, 'bus') and self.bus:
                message = self.bus.pop_filtered(Gst.MessageType.ERROR)
                if message:
                    err, debug = message.parse_error()
                    print(f"Error message: {err.message}")
                    print(f"Debug info: {debug}")
        
        except Exception as e:
            print(f"Error during diagnosis: {e}")
    
    def stop_pipeline(self) -> bool:
        """
        停止管道
        
        停止管道会将状态转换到NULL：
        - PLAYING -> PAUSED: 暂停数据流处理
        - PAUSED -> READY: 停止预处理，保持资源
        - READY -> NULL: 释放所有资源
        
        停止过程：
        1. 验证管道存在性
        2. 请求状态转换到NULL
        3. 等待状态转换完成
        4. 清理内部状态
        
        Returns:
            bool: 停止成功返回True，失败返回False
        """
        try:
            # 步骤1: 验证管道存在性
            if not self.pipeline:
                print("No pipeline to stop")
                return True  # 没有管道也算成功停止
            
            # 记录停止前的状态
            current_state = self.pipeline.get_state(0)[1]
            print(f"Stopping pipeline from state: {current_state}")
            
            # 步骤2: 请求状态转换到NULL
            # NULL状态会释放所有资源并重置管道
            print("Requesting pipeline state change to NULL...")
            ret = self.pipeline.set_state(Gst.State.NULL)
            
            # 步骤3: 检查状态转换结果
            if ret == Gst.StateChangeReturn.FAILURE:
                print("Failed to stop pipeline - state change failed")
                return False
            elif ret == Gst.StateChangeReturn.ASYNC:
                print("Pipeline state change is asynchronous, waiting...")
                # 等待状态转换完成（最多等待3秒）
                ret = self.pipeline.get_state(3 * Gst.SECOND)[0]
                if ret == Gst.StateChangeReturn.FAILURE:
                    print("Pipeline failed to reach NULL state")
                    return False
            
            # 步骤4: 清理内部状态
            self.is_playing = False
            if hasattr(self, 'start_time'):
                duration = time.time() - self.start_time
                print(f"Pipeline ran for {duration:.2f} seconds")
            
            # 验证最终状态
            final_state = self.pipeline.get_state(0)[1]
            print(f"Pipeline stopped successfully - final state: {final_state}")
            
            return True
        
        except Exception as e:
            print(f"Error stopping pipeline: {e}")
            self.error_count += 1
            return False
    
    def pause_pipeline(self) -> bool:
        """
        暂停管道
        
        暂停状态是GStreamer的重要中间状态：
        - 保持资源分配和连接
        - 停止数据流处理
        - 允许快速恢复到PLAYING状态
        - 支持精确的seek操作
        
        暂停过程：
        1. 验证管道存在性
        2. 请求状态转换到PAUSED
        3. 等待状态转换完成
        4. 更新内部状态
        
        Returns:
            bool: 暂停成功返回True，失败返回False
        """
        try:
            # 步骤1: 验证管道存在性
            if not self.pipeline:
                print("No pipeline to pause")
                return False
            
            # 记录暂停前的状态
            current_state = self.pipeline.get_state(0)[1]
            print(f"Pausing pipeline from state: {current_state}")
            
            # 步骤2: 请求状态转换到PAUSED
            # PAUSED状态保持资源但停止数据流
            print("Requesting pipeline state change to PAUSED...")
            ret = self.pipeline.set_state(Gst.State.PAUSED)
            
            # 步骤3: 检查状态转换结果
            if ret == Gst.StateChangeReturn.FAILURE:
                print("Failed to pause pipeline - state change failed")
                return False
            elif ret == Gst.StateChangeReturn.ASYNC:
                print("Pipeline state change is asynchronous, waiting...")
                # 等待状态转换完成（最多等待3秒）
                ret = self.pipeline.get_state(3 * Gst.SECOND)[0]
                if ret == Gst.StateChangeReturn.FAILURE:
                    print("Pipeline failed to reach PAUSED state")
                    return False
            elif ret == Gst.StateChangeReturn.SUCCESS:
                print("Pipeline state changed to PAUSED immediately")
            
            # 步骤4: 更新内部状态
            # 注意：暂停时不改变is_playing，因为可以快速恢复
            if hasattr(self, 'pause_time'):
                self.pause_time = time.time()
            
            # 验证最终状态
            final_state = self.pipeline.get_state(0)[1]
            print(f"Pipeline paused successfully - final state: {final_state}")
            
            return True
        
        except Exception as e:
            print(f"Error pausing pipeline: {e}")
            self.error_count += 1
            return False
    
    def seek(self, position_ns: int) -> bool:
        """跳转到指定位置（纳秒）"""
        try:
            if not self.pipeline:
                return False
            
            return self.pipeline.seek_simple(
                Gst.Format.TIME,
                Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,
                position_ns
            )
        
        except Exception as e:
            print(f"Error seeking: {e}")
            return False
    
    def get_position(self) -> Optional[int]:
        """获取当前播放位置（纳秒）"""
        try:
            if not self.pipeline:
                return None
            
            success, position = self.pipeline.query_position(Gst.Format.TIME)
            return position if success else None
        
        except Exception as e:
            print(f"Error getting position: {e}")
            return None
    
    def get_duration(self) -> Optional[int]:
        """获取总时长（纳秒）"""
        try:
            if not self.pipeline:
                return None
            
            success, duration = self.pipeline.query_duration(Gst.Format.TIME)
            return duration if success else None
        
        except Exception as e:
            print(f"Error getting duration: {e}")
            return None
    
    def _on_message(self, bus, message):
        """消息处理"""
        msg_type = message.type
        handler = self.message_handlers.get(msg_type)
        
        if handler:
            handler(message)
        else:
            # 处理其他消息类型
            if msg_type == Gst.MessageType.INFO:
                err, debug = message.parse_info()
                print(f"Info: {err.message}")
            elif msg_type == Gst.MessageType.WARNING:
                err, debug = message.parse_warning()
                print(f"Warning: {err.message}")
    
    def _handle_error(self, message):
        """处理错误消息"""
        err, debug = message.parse_error()
        print(f"Error: {err.message}")
        if debug:
            print(f"Debug info: {debug}")
        
        if self.error_callback:
            self.error_callback(err.message, debug)
        
        # 停止管道
        self.stop_pipeline()
    
    def _handle_eos(self, message):
        """处理流结束消息"""
        print("End of stream reached")
        
        if self.eos_callback:
            self.eos_callback()
        
        # 停止管道
        self.stop_pipeline()
    
    def _handle_state_changed(self, message):
        """处理状态变化消息"""
        if message.src == self.pipeline:
            old_state, new_state, pending_state = message.parse_state_changed()
            print(f"Pipeline state changed: {old_state.value_nick} -> {new_state.value_nick}")
            
            if self.state_change_callback:
                self.state_change_callback(old_state, new_state, pending_state)
    
    def _handle_buffering(self, message):
        """处理缓冲消息"""
        percent = message.parse_buffering()
        print(f"Buffering: {percent}%")
        
        # 当缓冲不足时暂停，缓冲充足时恢复播放
        if percent < 100:
            self.pipeline.set_state(Gst.State.PAUSED)
        else:
            self.pipeline.set_state(Gst.State.PLAYING)
    
    def _handle_stream_status(self, message):
        """处理流状态消息"""
        status_type, owner = message.parse_stream_status()
        print(f"Stream status: {status_type.value_nick} from {owner.get_name()}")
    
    def set_callbacks(self, state_change_cb=None, error_cb=None, eos_cb=None):
        """设置回调函数"""
        self.state_change_callback = state_change_cb
        self.error_callback = error_cb
        self.eos_callback = eos_cb
    
    def get_pipeline_info(self) -> Dict:
        """获取管道信息"""
        if not self.pipeline:
            return {}
        
        # 获取管道状态
        ret, state, pending = self.pipeline.get_state(Gst.CLOCK_TIME_NONE)
        
        # 获取位置和时长
        position = self.get_position()
        duration = self.get_duration()
        
        return {
            'name': self.pipeline.get_name(),
            'state': state.value_nick if ret == Gst.StateChangeReturn.SUCCESS else 'unknown',
            'pending_state': pending.value_nick if pending != Gst.State.VOID_PENDING else None,
            'position_ns': position,
            'duration_ns': duration,
            'position_sec': position / Gst.SECOND if position else None,
            'duration_sec': duration / Gst.SECOND if duration else None,
            'elements_count': len(self.elements),
            'elements': list(self.elements.keys())
        }
    
    def cleanup(self):
        """清理资源"""
        if self.pipeline:
            self.stop_pipeline()
            self.pipeline = None
        
        if self.bus:
            self.bus.remove_signal_watch()
            self.bus = None
        
        self.elements.clear()
        print("GStreamer core cleaned up")

# GStreamer核心演示
def demo_gstreamer_core():
    print("GStreamer Core Demo")
    print("===================")
    
    # 创建GStreamer核心管理器
    gst_core = GStreamerCore()
    
    # 设置回调函数
    def on_state_change(old_state, new_state, pending_state):
        print(f"State change callback: {old_state.value_nick} -> {new_state.value_nick}")
    
    def on_error(error_msg, debug_info):
        print(f"Error callback: {error_msg}")
    
    def on_eos():
        print("EOS callback: Stream ended")
    
    gst_core.set_callbacks(
        state_change_cb=on_state_change,
        error_cb=on_error,
        eos_cb=on_eos
    )
    
    # 创建管道
    if not gst_core.create_pipeline("demo-pipeline"):
        return
    
    # 创建一个简单的测试管道：videotestsrc -> autovideosink
    print("\n1. Creating test pipeline")
    
    # 添加视频测试源
    video_src = gst_core.add_element("videotestsrc", "video-source")
    if video_src:
        gst_core.set_element_property("video-source", "pattern", 0)  # SMPTE color bars
        gst_core.set_element_property("video-source", "num-buffers", 300)  # 300帧后结束
    
    # 添加视频转换器
    video_convert = gst_core.add_element("videoconvert", "video-convert")
    
    # 添加视频接收器
    video_sink = gst_core.add_element("autovideosink", "video-sink")
    
    # 连接元素
    print("\n2. Linking elements")
    if not gst_core.link_elements("video-source", "video-convert"):
        gst_core.cleanup()
        return
    
    if not gst_core.link_elements("video-convert", "video-sink"):
        gst_core.cleanup()
        return
    
    # 显示管道信息
    print("\n3. Pipeline information")
    pipeline_info = gst_core.get_pipeline_info()
    for key, value in pipeline_info.items():
        print(f"  {key}: {value}")
    
    # 启动管道
    print("\n4. Starting pipeline")
    if gst_core.start_pipeline():
        print("Pipeline is running...")
        
        # 运行一段时间
        start_time = time.time()
        while gst_core.is_playing and (time.time() - start_time) < 10:
            time.sleep(0.1)
            
            # 每秒显示一次位置信息
            if int(time.time() - start_time) % 2 == 0:
                position = gst_core.get_position()
                duration = gst_core.get_duration()
                if position is not None:
                    pos_sec = position / Gst.SECOND
                    dur_sec = duration / Gst.SECOND if duration else 0
                    print(f"Position: {pos_sec:.1f}s / {dur_sec:.1f}s")
                time.sleep(1)
    
    # 清理资源
    print("\n5. Cleaning up")
    gst_core.cleanup()

# 运行核心演示
if __name__ == "__main__":
    demo_gstreamer_core()

2. 核心组件详解

2.1 元素（Elements）系统

GStreamer的元素是处理管道中的基本构建块，每个元素都有特定的功能：

class ElementFactory:
    """元素工厂类"""
    
    def __init__(self):
        # 预定义的常用元素
        self.common_elements = {
            # 视频源
            'video_sources': {
                'videotestsrc': 'Video test pattern generator',
                'v4l2src': 'Video4Linux2 source',
                'filesrc': 'File source',
                'souphttpsrc': 'HTTP source',
                'rtspsrc': 'RTSP source',
                'udpsrc': 'UDP source'
            },
            
            # 音频源
            'audio_sources': {
                'audiotestsrc': 'Audio test signal generator',
                'alsasrc': 'ALSA audio source',
                'pulsesrc': 'PulseAudio source',
                'filesrc': 'File source'
            },
            
            # 视频编码器
            'video_encoders': {
                'x264enc': 'H.264 encoder',
                'x265enc': 'H.265 encoder',
                'vp8enc': 'VP8 encoder',
                'vp9enc': 'VP9 encoder',
                'theoraenc': 'Theora encoder'
            },
            
            # 音频编码器
            'audio_encoders': {
                'lamemp3enc': 'MP3 encoder',
                'faac': 'AAC encoder',
                'vorbisenc': 'Vorbis encoder',
                'opusenc': 'Opus encoder',
                'flacenc': 'FLAC encoder'
            },
            
            # 视频解码器
            'video_decoders': {
                'avdec_h264': 'H.264 decoder',
                'avdec_h265': 'H.265 decoder',
                'vp8dec': 'VP8 decoder',
                'vp9dec': 'VP9 decoder',
                'theoradec': 'Theora decoder'
            },
            
            # 音频解码器
            'audio_decoders': {
                'mpg123audiodec': 'MP3 decoder',
                'faad': 'AAC decoder',
                'vorbisdec': 'Vorbis decoder',
                'opusdec': 'Opus decoder',
                'flacdec': 'FLAC decoder'
            },
            
            # 复用器
            'muxers': {
                'mp4mux': 'MP4 muxer',
                'matroskamux': 'Matroska muxer',
                'avimux': 'AVI muxer',
                'oggmux': 'Ogg muxer',
                'flvmux': 'FLV muxer'
            },
            
            # 解复用器
            'demuxers': {
                'qtdemux': 'QuickTime/MP4 demuxer',
                'matroskademux': 'Matroska demuxer',
                'avidemux': 'AVI demuxer',
                'oggdemux': 'Ogg demuxer',
                'flvdemux': 'FLV demuxer'
            },
            
            # 过滤器和转换器
            'filters': {
                'videoconvert': 'Video format converter',
                'videoscale': 'Video scaler',
                'videorate': 'Video framerate adjuster',
                'audioconvert': 'Audio format converter',
                'audioresample': 'Audio resampler',
                'capsfilter': 'Caps filter'
            },
            
            # 接收器
            'sinks': {
                'autovideosink': 'Automatic video sink',
                'autoaudiosink': 'Automatic audio sink',
                'filesink': 'File sink',
                'udpsink': 'UDP sink',
                'tcpserversink': 'TCP server sink',
                'appsink': 'Application sink'
            }
        }
    
    def get_element_info(self, factory_name: str) -> Optional[ElementInfo]:
        """获取元素信息"""
        try:
            factory = Gst.ElementFactory.find(factory_name)
            if not factory:
                return None
            
            # 创建临时元素以获取信息
            element = factory.create(None)
            if not element:
                return None
            
            # 获取Pad模板
            src_pads = []
            sink_pads = []
            
            for pad_template in factory.get_static_pad_templates():
                if pad_template.direction == Gst.PadDirection.SRC:
                    src_pads.append(pad_template.name_template)
                elif pad_template.direction == Gst.PadDirection.SINK:
                    sink_pads.append(pad_template.name_template)
            
            # 获取属性
            properties = {}
            for prop in element.list_properties():
                prop_name = prop.name
                prop_type = prop.value_type.name
                default_value = element.get_property(prop_name)
                
                properties[prop_name] = {
                    'type': prop_type,
                    'default': default_value,
                    'description': prop.blurb
                }
            
            return ElementInfo(
                name=factory_name,
                element_type=self._classify_element(factory_name),
                factory_name=factory_name,
                description=factory.get_description(),
                src_pads=src_pads,
                sink_pads=sink_pads,
                properties=properties
            )
        
        except Exception as e:
            print(f"Error getting element info for {factory_name}: {e}")
            return None
    
    def _classify_element(self, factory_name: str) -> ElementType:
        """分类元素类型"""
        for category, elements in self.common_elements.items():
            if factory_name in elements:
                if 'source' in category:
                    return ElementType.SOURCE
                elif 'encoder' in category or 'decoder' in category:
                    return ElementType.CODEC
                elif 'mux' in category:
                    return ElementType.MUXER if 'demux' not in category else ElementType.DEMUXER
                elif 'filter' in category or 'convert' in category:
                    return ElementType.CONVERTER
                elif 'sink' in category:
                    return ElementType.SINK
        
        # 默认分类
        if 'src' in factory_name:
            return ElementType.SOURCE
        elif 'sink' in factory_name:
            return ElementType.SINK
        elif 'enc' in factory_name:
            return ElementType.CODEC
        elif 'dec' in factory_name:
            return ElementType.CODEC
        elif 'mux' in factory_name:
            return ElementType.MUXER
        elif 'demux' in factory_name:
            return ElementType.DEMUXER
        else:
            return ElementType.FILTER
    
    def list_available_elements(self, element_type: ElementType = None) -> List[str]:
        """列出可用元素"""
        available_elements = []
        
        # 获取所有已注册的元素工厂
        registry = Gst.Registry.get()
        factories = registry.get_feature_list(Gst.ElementFactory)
        
        for factory in factories:
            factory_name = factory.get_name()
            
            if element_type:
                if self._classify_element(factory_name) == element_type:
                    available_elements.append(factory_name)
            else:
                available_elements.append(factory_name)
        
        return sorted(available_elements)
    
    def find_elements_by_caps(self, caps_string: str, direction: PadDirection) -> List[str]:
        """根据Caps查找兼容的元素"""
        compatible_elements = []
        
        try:
            caps = Gst.Caps.from_string(caps_string)
            if not caps:
                return compatible_elements
            
            registry = Gst.Registry.get()
            factories = registry.get_feature_list(Gst.ElementFactory)
            
            for factory in factories:
                # 检查元素的Pad模板
                for pad_template in factory.get_static_pad_templates():
                    if pad_template.direction.value_nick == direction.value:
                        template_caps = pad_template.get_caps()
                        if template_caps.can_intersect(caps):
                            compatible_elements.append(factory.get_name())
                            break
        
        except Exception as e:
            print(f"Error finding elements by caps: {e}")
        
        return compatible_elements
    
    def get_recommended_pipeline(self, input_format: str, output_format: str) -> List[str]:
        """获取推荐的管道配置"""
        pipeline_elements = []
        
        # 简化的管道推荐逻辑
        format_mappings = {
            'mp4': {
                'demuxer': 'qtdemux',
                'muxer': 'mp4mux',
                'video_decoder': 'avdec_h264',
                'audio_decoder': 'faad',
                'video_encoder': 'x264enc',
                'audio_encoder': 'faac'
            },
            'mkv': {
                'demuxer': 'matroskademux',
                'muxer': 'matroskamux',
                'video_decoder': 'avdec_h264',
                'audio_decoder': 'vorbisdec',
                'video_encoder': 'x264enc',
                'audio_encoder': 'vorbisenc'
            },
            'webm': {
                'demuxer': 'matroskademux',
                'muxer': 'webmmux',
                'video_decoder': 'vp8dec',
                'audio_decoder': 'vorbisdec',
                'video_encoder': 'vp8enc',
                'audio_encoder': 'vorbisenc'
            }
        }
        
        input_config = format_mappings.get(input_format.lower())
        output_config = format_mappings.get(output_format.lower())
        
        if input_config and output_config:
            # 构建转码管道
            pipeline_elements = [
                'filesrc',
                input_config['demuxer'],
                input_config['video_decoder'],
                'videoconvert',
                output_config['video_encoder'],
                input_config['audio_decoder'],
                'audioconvert',
                output_config['audio_encoder'],
                output_config['muxer'],
                'filesink'
            ]
        
        return pipeline_elements

# 元素工厂演示
def demo_element_factory():
    print("Element Factory Demo")
    print("====================")
    
    factory = ElementFactory()
    
    # 1. 列出不同类型的元素
    print("\n1. Available Elements by Type")
    for element_type in ElementType:
        elements = factory.list_available_elements(element_type)
        print(f"  {element_type.value}: {len(elements)} elements")
        if elements:
            print(f"    Examples: {', '.join(elements[:5])}")
    
    # 2. 获取特定元素的详细信息
    print("\n2. Element Information")
    test_elements = ['videotestsrc', 'x264enc', 'mp4mux']
    
    for element_name in test_elements:
        info = factory.get_element_info(element_name)
        if info:
            print(f"\n  Element: {info.name}")
            print(f"    Type: {info.element_type.value}")
            print(f"    Description: {info.description}")
            print(f"    Source pads: {info.src_pads}")
            print(f"    Sink pads: {info.sink_pads}")
            print(f"    Properties: {len(info.properties)} available")
            
            # 显示几个重要属性
            important_props = ['bitrate', 'width', 'height', 'framerate', 'pattern']
            for prop_name in important_props:
                if prop_name in info.properties:
                    prop_info = info.properties[prop_name]
                    print(f"      {prop_name}: {prop_info['type']} (default: {prop_info['default']})")
    
    # 3. 根据Caps查找兼容元素
    print("\n3. Finding Compatible Elements")
    test_caps = [
        "video/x-raw",
        "video/x-h264",
        "audio/x-raw",
        "audio/mpeg"
    ]
    
    for caps_string in test_caps:
        print(f"\n  Caps: {caps_string}")
        
        # 查找源元素
        src_elements = factory.find_elements_by_caps(caps_string, PadDirection.SRC)
        print(f"    Source elements: {', '.join(src_elements[:5])}")
        
        # 查找接收元素
        sink_elements = factory.find_elements_by_caps(caps_string, PadDirection.SINK)
        print(f"    Sink elements: {', '.join(sink_elements[:5])}")
    
    # 4. 获取推荐管道
    print("\n4. Recommended Pipelines")
    format_pairs = [('mp4', 'webm'), ('mkv', 'mp4'), ('webm', 'mkv')]
    
    for input_fmt, output_fmt in format_pairs:
        pipeline = factory.get_recommended_pipeline(input_fmt, output_fmt)
        print(f"\n  {input_fmt.upper()} -> {output_fmt.upper()}:")
        print(f"    Pipeline: {' -> '.join(pipeline)}")

# 运行元素工厂演示
if __name__ == "__main__":
    demo_element_factory()

2.2 Pad和Caps系统

Pad是元素之间连接的接口，Caps定义了数据流的格式和属性：

class PadManager:
    """Pad管理器"""
    
    def __init__(self):
        self.pad_probes = {}  # Pad探针
        self.pad_callbacks = {}  # Pad回调
    
    def add_pad_probe(self, element: Gst.Element, pad_name: str, 
                     probe_type: Gst.PadProbeType, callback: Callable) -> int:
        """添加Pad探针"""
        try:
            pad = element.get_static_pad(pad_name)
            if not pad:
                print(f"Pad '{pad_name}' not found on element {element.get_name()}")
                return 0
            
            probe_id = pad.add_probe(probe_type, callback)
            
            probe_key = f"{element.get_name()}:{pad_name}"
            self.pad_probes[probe_key] = probe_id
            
            print(f"Added probe to {probe_key} (ID: {probe_id})")
            return probe_id
        
        except Exception as e:
            print(f"Error adding pad probe: {e}")
            return 0
    
    def remove_pad_probe(self, element: Gst.Element, pad_name: str) -> bool:
        """移除Pad探针"""
        try:
            probe_key = f"{element.get_name()}:{pad_name}"
            probe_id = self.pad_probes.get(probe_key)
            
            if not probe_id:
                print(f"No probe found for {probe_key}")
                return False
            
            pad = element.get_static_pad(pad_name)
            if pad:
                pad.remove_probe(probe_id)
                del self.pad_probes[probe_key]
                print(f"Removed probe from {probe_key}")
                return True
            
            return False
        
        except Exception as e:
            print(f"Error removing pad probe: {e}")
            return False
    
    def get_pad_caps(self, element: Gst.Element, pad_name: str) -> Optional[str]:
        """获取Pad的Caps"""
        try:
            pad = element.get_static_pad(pad_name)
            if not pad:
                return None
            
            caps = pad.get_current_caps()
            if not caps:
                caps = pad.query_caps(None)
            
            return caps.to_string() if caps else None
        
        except Exception as e:
            print(f"Error getting pad caps: {e}")
            return None
    
    def set_pad_caps(self, element: Gst.Element, pad_name: str, caps_string: str) -> bool:
        """设置Pad的Caps"""
        try:
            pad = element.get_static_pad(pad_name)
            if not pad:
                return False
            
            caps = Gst.Caps.from_string(caps_string)
            if not caps:
                print(f"Invalid caps string: {caps_string}")
                return False
            
            # 通过capsfilter设置caps
            return pad.set_caps(caps)
        
        except Exception as e:
            print(f"Error setting pad caps: {e}")
            return False
    
    def link_pads_with_caps(self, src_element: Gst.Element, src_pad: str,
                           sink_element: Gst.Element, sink_pad: str,
                           caps_string: str) -> bool:
        """使用指定Caps连接Pad"""
        try:
            src_pad_obj = src_element.get_static_pad(src_pad)
            sink_pad_obj = sink_element.get_static_pad(sink_pad)
            
            if not src_pad_obj or not sink_pad_obj:
                return False
            
            caps = Gst.Caps.from_string(caps_string)
            if not caps:
                return False
            
            # 检查caps兼容性
            src_caps = src_pad_obj.query_caps(None)
            sink_caps = sink_pad_obj.query_caps(None)
            
            if not src_caps.can_intersect(caps) or not sink_caps.can_intersect(caps):
                print(f"Caps '{caps_string}' not compatible with pads")
                return False
            
            # 连接pad
            result = src_pad_obj.link(sink_pad_obj)
            return result == Gst.PadLinkReturn.OK
        
        except Exception as e:
            print(f"Error linking pads with caps: {e}")
            return False

class CapsManager:
    """Caps管理器"""
    
    def __init__(self):
        # 常用的Caps模板
        self.caps_templates = {
            'video_raw': 'video/x-raw,format={format},width={width},height={height},framerate={framerate}/1',
            'video_h264': 'video/x-h264,profile={profile},level={level},width={width},height={height},framerate={framerate}/1',
            'video_vp8': 'video/x-vp8,width={width},height={height},framerate={framerate}/1',
            'audio_raw': 'audio/x-raw,format={format},rate={rate},channels={channels},layout=interleaved',
            'audio_mp3': 'audio/mpeg,mpegversion=1,layer=3,rate={rate},channels={channels}',
            'audio_aac': 'audio/mpeg,mpegversion=4,rate={rate},channels={channels}'
        }
    
    def create_video_caps(self, format_type: str = 'raw', width: int = 1920, 
                         height: int = 1080, framerate: int = 30, **kwargs) -> str:
        """创建视频Caps"""
        if format_type == 'raw':
            format_name = kwargs.get('format', 'I420')
            return self.caps_templates['video_raw'].format(
                format=format_name, width=width, height=height, framerate=framerate
            )
        elif format_type == 'h264':
            profile = kwargs.get('profile', 'high')
            level = kwargs.get('level', '4.0')
            return self.caps_templates['video_h264'].format(
                profile=profile, level=level, width=width, height=height, framerate=framerate
            )
        elif format_type == 'vp8':
            return self.caps_templates['video_vp8'].format(
                width=width, height=height, framerate=framerate
            )
        else:
            return f"video/x-{format_type},width={width},height={height},framerate={framerate}/1"
    
    def create_audio_caps(self, format_type: str = 'raw', rate: int = 44100, 
                         channels: int = 2, **kwargs) -> str:
        """创建音频Caps"""
        if format_type == 'raw':
            format_name = kwargs.get('format', 'S16LE')
            return self.caps_templates['audio_raw'].format(
                format=format_name, rate=rate, channels=channels
            )
        elif format_type == 'mp3':
            return self.caps_templates['audio_mp3'].format(
                rate=rate, channels=channels
            )
        elif format_type == 'aac':
            return self.caps_templates['audio_aac'].format(
                rate=rate, channels=channels
            )
        else:
            return f"audio/x-{format_type},rate={rate},channels={channels}"
    
    def parse_caps(self, caps_string: str) -> Dict[str, Any]:
        """解析Caps字符串"""
        caps_info = {
            'media_type': None,
            'format': None,
            'properties': {}
        }
        
        try:
            caps = Gst.Caps.from_string(caps_string)
            if not caps or caps.get_size() == 0:
                return caps_info
            
            structure = caps.get_structure(0)
            if not structure:
                return caps_info
            
            # 获取媒体类型
            caps_info['media_type'] = structure.get_name()
            
            # 解析属性
            for i in range(structure.n_fields()):
                field_name = structure.nth_field_name(i)
                field_value = structure.get_value(field_name)
                caps_info['properties'][field_name] = field_value
            
            # 提取常用属性
            if 'format' in caps_info['properties']:
                caps_info['format'] = caps_info['properties']['format']
            
            return caps_info
        
        except Exception as e:
            print(f"Error parsing caps: {e}")
            return caps_info
    
    def caps_intersect(self, caps1_string: str, caps2_string: str) -> Optional[str]:
        """计算两个Caps的交集"""
        try:
            caps1 = Gst.Caps.from_string(caps1_string)
            caps2 = Gst.Caps.from_string(caps2_string)
            
            if not caps1 or not caps2:
                return None
            
            intersection = caps1.intersect(caps2)
            if intersection.is_empty():
                return None
            
            return intersection.to_string()
        
        except Exception as e:
            print(f"Error calculating caps intersection: {e}")
            return None
    
    def caps_can_intersect(self, caps1_string: str, caps2_string: str) -> bool:
        """检查两个Caps是否可以相交"""
        try:
            caps1 = Gst.Caps.from_string(caps1_string)
            caps2 = Gst.Caps.from_string(caps2_string)
            
            if not caps1 or not caps2:
                return False
            
            return caps1.can_intersect(caps2)
        
        except Exception as e:
            print(f"Error checking caps intersection: {e}")
            return False
    
    def get_compatible_formats(self, base_caps: str, available_formats: List[str]) -> List[str]:
        """获取兼容的格式"""
        compatible = []
        
        for format_caps in available_formats:
            if self.caps_can_intersect(base_caps, format_caps):
                compatible.append(format_caps)
        
        return compatible

# Pad和Caps演示
def demo_pad_caps():
    print("Pad and Caps Demo")
    print("=================")
    
    pad_manager = PadManager()
    caps_manager = CapsManager()
    
    # 1. 创建和解析Caps
    print("\n1. Creating and Parsing Caps")
    
    # 创建不同类型的Caps
    video_caps = caps_manager.create_video_caps('h264', 1280, 720, 25, profile='main')
    audio_caps = caps_manager.create_audio_caps('aac', 48000, 2)
    
    print(f"Video caps: {video_caps}")
    print(f"Audio caps: {audio_caps}")
    
    # 解析Caps
    video_info = caps_manager.parse_caps(video_caps)
    audio_info = caps_manager.parse_caps(audio_caps)
    
    print(f"\nVideo caps info:")
    print(f"  Media type: {video_info['media_type']}")
    print(f"  Properties: {video_info['properties']}")
    
    print(f"\nAudio caps info:")
    print(f"  Media type: {audio_info['media_type']}")
    print(f"  Properties: {audio_info['properties']}")
    
    # 2. Caps兼容性检查
    print("\n2. Caps Compatibility")
    
    test_caps = [
        caps_manager.create_video_caps('raw', 1920, 1080, 30),
        caps_manager.create_video_caps('h264', 1920, 1080, 30),
        caps_manager.create_video_caps('h264', 1280, 720, 25),
        "video/x-raw,width=1920,height=1080"
    ]
    
    base_caps = "video/x-raw,width=[1,4096],height=[1,4096],framerate=[1/1,60/1]"
    
    print(f"Base caps: {base_caps}")
    print("\nCompatibility check:")
    
    for i, caps in enumerate(test_caps):
        compatible = caps_manager.caps_can_intersect(base_caps, caps)
        intersection = caps_manager.caps_intersect(base_caps, caps)
        
        print(f"  {i+1}. {caps[:50]}...")
        print(f"     Compatible: {compatible}")
        if intersection:
            print(f"     Intersection: {intersection[:50]}...")
    
    # 3. 创建简单管道测试Pad
    print("\n3. Pad Testing with Pipeline")
    
    gst_core = GStreamerCore()
    if gst_core.create_pipeline("pad-test-pipeline"):
        # 添加元素
        videotestsrc = gst_core.add_element("videotestsrc", "video-src")
        capsfilter = gst_core.add_element("capsfilter", "caps-filter")
        fakesink = gst_core.add_element("fakesink", "fake-sink")
        
        if videotestsrc and capsfilter and fakesink:
            # 设置caps filter
            test_caps = caps_manager.create_video_caps('raw', 640, 480, 15, format='RGB')
            gst_core.set_element_property("caps-filter", "caps", Gst.Caps.from_string(test_caps))
            
            # 连接元素
            if (gst_core.link_elements("video-src", "caps-filter") and
                gst_core.link_elements("caps-filter", "fake-sink")):
                
                print("Pipeline created successfully")
                
                # 获取Pad信息
                src_caps = pad_manager.get_pad_caps(videotestsrc, "src")
                filter_sink_caps = pad_manager.get_pad_caps(capsfilter, "sink")
                filter_src_caps = pad_manager.get_pad_caps(capsfilter, "src")
                
                print(f"\nPad caps:")
                print(f"  videotestsrc src: {src_caps[:80] if src_caps else 'None'}...")
                print(f"  capsfilter sink: {filter_sink_caps[:80] if filter_sink_caps else 'None'}...")
                print(f"  capsfilter src: {filter_src_caps[:80] if filter_src_caps else 'None'}...")
        
        gst_core.cleanup()
    
    print("\nPad and Caps demo completed")

# 运行Pad和Caps演示
if __name__ == "__main__":
    demo_pad_caps()

3. 管道构建器

3.1 高级管道构建

为了简化复杂管道的构建，我们实现一个高级的管道构建器：

class PipelineBuilder:
    """管道构建器"""
    
    def __init__(self, name: str = "custom-pipeline"):
        self.gst_core = GStreamerCore()
        self.pipeline_name = name
        self.element_configs = []
        self.link_configs = []
        self.property_configs = []
        self.caps_configs = []
        
        # 预定义的管道模板
        self.pipeline_templates = {
            'video_player': self._video_player_template,
            'audio_player': self._audio_player_template,
            'video_transcoder': self._video_transcoder_template,
            'rtmp_streamer': self._rtmp_streamer_template,
            'webcam_capture': self._webcam_capture_template,
            'screen_recorder': self._screen_recorder_template
        }
    
    def add_element(self, factory_name: str, element_name: str, properties: Dict[str, Any] = None) -> 'PipelineBuilder':
        """添加元素配置"""
        self.element_configs.append({
            'factory_name': factory_name,
            'element_name': element_name,
            'properties': properties or {}
        })
        return self
    
    def link(self, src_element: str, sink_element: str, 
            src_pad: str = None, sink_pad: str = None, caps: str = None) -> 'PipelineBuilder':
        """添加连接配置"""
        self.link_configs.append({
            'src_element': src_element,
            'sink_element': sink_element,
            'src_pad': src_pad,
            'sink_pad': sink_pad,
            'caps': caps
        })
        return self
    
    def set_property(self, element_name: str, property_name: str, value: Any) -> 'PipelineBuilder':
        """添加属性配置"""
        self.property_configs.append({
            'element_name': element_name,
            'property_name': property_name,
            'value': value
        })
        return self
    
    def set_caps(self, element_name: str, caps: str) -> 'PipelineBuilder':
        """添加Caps配置"""
        self.caps_configs.append({
            'element_name': element_name,
            'caps': caps
        })
        return self
    
    def build(self) -> Optional[GStreamerCore]:
        """构建管道"""
        try:
            # 创建管道
            if not self.gst_core.create_pipeline(self.pipeline_name):
                return None
            
            # 添加所有元素
            print(f"Building pipeline '{self.pipeline_name}'...")
            for config in self.element_configs:
                element = self.gst_core.add_element(
                    config['factory_name'], 
                    config['element_name']
                )
                
                if not element:
                    print(f"Failed to add element: {config['element_name']}")
                    return None
                
                # 设置元素属性
                for prop_name, prop_value in config['properties'].items():
                    self.gst_core.set_element_property(
                        config['element_name'], prop_name, prop_value
                    )
            
            # 设置额外属性
            for config in self.property_configs:
                self.gst_core.set_element_property(
                    config['element_name'],
                    config['property_name'],
                    config['value']
                )
            
            # 设置Caps
            for config in self.caps_configs:
                element = self.gst_core.elements.get(config['element_name'])
                if element and 'caps' in [prop.name for prop in element.list_properties()]:
                    caps = Gst.Caps.from_string(config['caps'])
                    if caps:
                        element.set_property('caps', caps)
            
            # 连接元素
            for config in self.link_configs:
                if config['caps']:
                    # 使用capsfilter进行连接
                    capsfilter_name = f"capsfilter_{len(self.gst_core.elements)}"
                    capsfilter = self.gst_core.add_element("capsfilter", capsfilter_name)
                    if capsfilter:
                        caps = Gst.Caps.from_string(config['caps'])
                        capsfilter.set_property('caps', caps)
                        
                        # 连接：src -> capsfilter -> sink
                        if not (self.gst_core.link_elements(config['src_element'], capsfilter_name) and
                               self.gst_core.link_elements(capsfilter_name, config['sink_element'])):
                            print(f"Failed to link with caps: {config['src_element']} -> {config['sink_element']}")
                            return None
                else:
                    # 直接连接
                    if not self.gst_core.link_elements(
                        config['src_element'],
                        config['sink_element'],
                        config['src_pad'],
                        config['sink_pad']
                    ):
                        print(f"Failed to link: {config['src_element']} -> {config['sink_element']}")
                        return None
            
            print(f"Pipeline '{self.pipeline_name}' built successfully")
            return self.gst_core
        
        except Exception as e:
            print(f"Error building pipeline: {e}")
            return None
    
    def from_template(self, template_name: str, **kwargs) -> 'PipelineBuilder':
        """从模板创建管道"""
        template_func = self.pipeline_templates.get(template_name)
        if template_func:
            return template_func(**kwargs)
        else:
            print(f"Template '{template_name}' not found")
            return self
    
    def _video_player_template(self, input_file: str, **kwargs) -> 'PipelineBuilder':
        """视频播放器模板"""
        return (self
                .add_element("filesrc", "file-source", {'location': input_file})
                .add_element("decodebin", "decoder")
                .add_element("videoconvert", "video-convert")
                .add_element("videoscale", "video-scale")
                .add_element("autovideosink", "video-sink")
                .add_element("audioconvert", "audio-convert")
                .add_element("audioresample", "audio-resample")
                .add_element("autoaudiosink", "audio-sink"))
    
    def _audio_player_template(self, input_file: str, **kwargs) -> 'PipelineBuilder':
        """音频播放器模板"""
        return (self
                .add_element("filesrc", "file-source", {'location': input_file})
                .add_element("decodebin", "decoder")
                .add_element("audioconvert", "audio-convert")
                .add_element("audioresample", "audio-resample")
                .add_element("autoaudiosink", "audio-sink"))
    
    def _video_transcoder_template(self, input_file: str, output_file: str, 
                                  video_codec: str = 'h264', audio_codec: str = 'aac', **kwargs) -> 'PipelineBuilder':
        """视频转码器模板"""
        # 编码器映射
        video_encoders = {
            'h264': 'x264enc',
            'h265': 'x265enc',
            'vp8': 'vp8enc',
            'vp9': 'vp9enc'
        }
        
        audio_encoders = {
            'aac': 'faac',
            'mp3': 'lamemp3enc',
            'vorbis': 'vorbisenc',
            'opus': 'opusenc'
        }
        
        # 复用器映射
        output_ext = output_file.split('.')[-1].lower()
        muxers = {
            'mp4': 'mp4mux',
            'mkv': 'matroskamux',
            'webm': 'webmmux',
            'avi': 'avimux'
        }
        
        video_enc = video_encoders.get(video_codec, 'x264enc')
        audio_enc = audio_encoders.get(audio_codec, 'faac')
        muxer = muxers.get(output_ext, 'mp4mux')
        
        return (self
                .add_element("filesrc", "file-source", {'location': input_file})
                .add_element("decodebin", "decoder")
                .add_element("videoconvert", "video-convert")
                .add_element("videoscale", "video-scale")
                .add_element(video_enc, "video-encoder")
                .add_element("audioconvert", "audio-convert")
                .add_element("audioresample", "audio-resample")
                .add_element(audio_enc, "audio-encoder")
                .add_element(muxer, "muxer")
                .add_element("filesink", "file-sink", {'location': output_file}))
    
    def _rtmp_streamer_template(self, rtmp_url: str, video_device: str = '/dev/video0', **kwargs) -> 'PipelineBuilder':
        """RTMP推流模板"""
        return (self
                .add_element("v4l2src", "video-source", {'device': video_device})
                .add_element("videoconvert", "video-convert")
                .add_element("videoscale", "video-scale")
                .add_element("x264enc", "video-encoder", {
                    'bitrate': kwargs.get('video_bitrate', 2000),
                    'tune': 'zerolatency'
                })
                .add_element("alsasrc", "audio-source")
                .add_element("audioconvert", "audio-convert")
                .add_element("audioresample", "audio-resample")
                .add_element("faac", "audio-encoder", {
                    'bitrate': kwargs.get('audio_bitrate', 128000)
                })
                .add_element("flvmux", "muxer")
                .add_element("rtmpsink", "rtmp-sink", {'location': rtmp_url}))
    
    def _webcam_capture_template(self, output_file: str, video_device: str = '/dev/video0', **kwargs) -> 'PipelineBuilder':
        """摄像头录制模板"""
        width = kwargs.get('width', 1280)
        height = kwargs.get('height', 720)
        framerate = kwargs.get('framerate', 30)
        
        video_caps = f"video/x-raw,width={width},height={height},framerate={framerate}/1"
        
        return (self
                .add_element("v4l2src", "video-source", {'device': video_device})
                .add_element("videoconvert", "video-convert")
                .add_element("videoscale", "video-scale")
                .add_element("x264enc", "video-encoder")
                .add_element("mp4mux", "muxer")
                .add_element("filesink", "file-sink", {'location': output_file})
                .set_caps("video-source", video_caps))
    
    def _screen_recorder_template(self, output_file: str, **kwargs) -> 'PipelineBuilder':
        """屏幕录制模板"""
        x = kwargs.get('x', 0)
        y = kwargs.get('y', 0)
        width = kwargs.get('width', 1920)
        height = kwargs.get('height', 1080)
        
        return (self
                .add_element("ximagesrc", "screen-source", {
                    'startx': x,
                    'starty': y,
                    'endx': x + width,
                    'endy': y + height
                })
                .add_element("videoconvert", "video-convert")
                .add_element("videorate", "video-rate")
                .add_element("x264enc", "video-encoder")
                .add_element("mp4mux", "muxer")
                .add_element("filesink", "file-sink", {'location': output_file}))
    
    def get_available_templates(self) -> List[str]:
        """获取可用模板列表"""
        return list(self.pipeline_templates.keys())
    
    def describe_template(self, template_name: str) -> str:
        """描述模板功能"""
        descriptions = {
            'video_player': 'Play video files with automatic format detection',
            'audio_player': 'Play audio files with automatic format detection',
            'video_transcoder': 'Transcode video files between different formats',
            'rtmp_streamer': 'Stream video from webcam to RTMP server',
            'webcam_capture': 'Capture video from webcam to file',
            'screen_recorder': 'Record screen content to video file'
        }
        return descriptions.get(template_name, 'No description available')

# 管道构建器演示
def demo_pipeline_builder():
    print("Pipeline Builder Demo")
    print("=====================")
    
    # 1. 显示可用模板
    print("\n1. Available Pipeline Templates")
    builder = PipelineBuilder()
    templates = builder.get_available_templates()
    
    for template in templates:
        description = builder.describe_template(template)
        print(f"  {template}: {description}")
    
    # 2. 使用模板构建简单播放器
    print("\n2. Building Video Test Player")
    
    # 创建视频测试播放器（使用videotestsrc代替文件）
    player_builder = (PipelineBuilder("test-video-player")
                     .add_element("videotestsrc", "video-source", {
                         'pattern': 0,  # SMPTE color bars
                         'num-buffers': 300  # 10秒 @ 30fps
                     })
                     .add_element("videoconvert", "video-convert")
                     .add_element("videoscale", "video-scale")
                     .add_element("autovideosink", "video-sink")
                     .link("video-source", "video-convert")
                     .link("video-convert", "video-scale")
                     .link("video-scale", "video-sink"))
    
    # 构建管道
    player_core = player_builder.build()
    if player_core:
        print("Test video player built successfully")
        
        # 显示管道信息
        pipeline_info = player_core.get_pipeline_info()
        print(f"Pipeline info: {pipeline_info['elements_count']} elements")
        print(f"Elements: {', '.join(pipeline_info['elements'])}")
        
        # 启动播放（短时间）
        if player_core.start_pipeline():
            print("Playing test video for 3 seconds...")
            time.sleep(3)
        
        player_core.cleanup()
    
    # 3. 构建自定义转码管道
    print("\n3. Building Custom Transcoding Pipeline")
    
    # 创建音频转码管道（audiotestsrc -> encoder -> filesink）
    transcoder_builder = (PipelineBuilder("audio-transcoder")
                         .add_element("audiotestsrc", "audio-source", {
                             'num-buffers': 1000,  # 约23秒
                             'freq': 440  # A4音符
                         })
                         .add_element("audioconvert", "audio-convert")
                         .add_element("audioresample", "audio-resample")
                         .add_element("lamemp3enc", "mp3-encoder", {
                             'bitrate': 128
                         })
                         .add_element("filesink", "file-sink", {
                             'location': 'test_output.mp3'
                         })
                         .link("audio-source", "audio-convert")
                         .link("audio-convert", "audio-resample")
                         .link("audio-resample", "mp3-encoder")
                         .link("mp3-encoder", "file-sink"))
    
    transcoder_core = transcoder_builder.build()
    if transcoder_core:
        print("Audio transcoder built successfully")
        
        # 显示管道信息
        pipeline_info = transcoder_core.get_pipeline_info()
        print(f"Pipeline: {' -> '.join(pipeline_info['elements'])}")
        
        transcoder_core.cleanup()
    
    # 4. 使用Caps约束构建管道
    print("\n4. Building Pipeline with Caps Constraints")
    
    caps_builder = (PipelineBuilder("caps-constrained-pipeline")
                   .add_element("videotestsrc", "video-source")
                   .add_element("capsfilter", "caps-filter")
                   .add_element("videoconvert", "video-convert")
                   .add_element("fakesink", "fake-sink")
                   .set_caps("caps-filter", "video/x-raw,width=320,height=240,framerate=15/1")
                   .link("video-source", "caps-filter")
                   .link("caps-filter", "video-convert")
                   .link("video-convert", "fake-sink"))
    
    caps_core = caps_builder.build()
    if caps_core:
        print("Caps-constrained pipeline built successfully")
        caps_core.cleanup()
    
    print("\nPipeline Builder demo completed")

# 运行管道构建器演示
if __name__ == "__main__":
    demo_pipeline_builder()

4. 媒体处理应用

4.1 实时视频处理

GStreamer支持实时视频处理，包括滤镜、特效和分析：

class VideoProcessor:
    """视频处理器"""
    
    def __init__(self):
        self.gst_core = GStreamerCore()
        self.processing_elements = {
            'filters': {
                'blur': 'gaussianblur',
                'edge': 'edgetv',
                'emboss': 'videoflip',
                'colorbalance': 'videobalance',
                'gamma': 'gamma',
                'contrast': 'videobalance'
            },
            'effects': {
                'aging': 'agingtv',
                'dice': 'dicetv',
                'warp': 'warptv',
                'radioac': 'radioactv',
                'revtv': 'revtv',
                'streaktv': 'streaktv'
            },
            'analysis': {
                'motion': 'motioncells',
                'face': 'facedetect',
                'object': 'opencv'
            }
        }
    
    def create_filter_pipeline(self, input_source: str, output_sink: str, 
                              filters: List[str], filter_params: Dict[str, Dict] = None) -> bool:
        """创建滤镜处理管道"""
        try:
            if not self.gst_core.create_pipeline("video-filter-pipeline"):
                return False
            
            filter_params = filter_params or {}
            
            # 添加输入源
            if input_source.startswith('/dev/video'):
                # 摄像头输入
                self.gst_core.add_element("v4l2src", "video-source")
                self.gst_core.set_element_property("video-source", "device", input_source)
            elif input_source == "test":
                # 测试源
                self.gst_core.add_element("videotestsrc", "video-source")
            else:
                # 文件输入
                self.gst_core.add_element("filesrc", "file-source")
                self.gst_core.set_element_property("file-source", "location", input_source)
                self.gst_core.add_element("decodebin", "decoder")
            
            # 添加格式转换
            self.gst_core.add_element("videoconvert", "video-convert-in")
            
            # 添加滤镜链
            previous_element = "video-convert-in"
            for i, filter_name in enumerate(filters):
                # 查找滤镜元素
                filter_element = None
                for category, elements in self.processing_elements.items():
                    if filter_name in elements:
                        filter_element = elements[filter_name]
                        break
                
                if not filter_element:
                    print(f"Unknown filter: {filter_name}")
                    continue
                
                element_name = f"filter-{i}-{filter_name}"
                self.gst_core.add_element(filter_element, element_name)
                
                # 设置滤镜参数
                if filter_name in filter_params:
                    for prop, value in filter_params[filter_name].items():
                        self.gst_core.set_element_property(element_name, prop, value)
                
                # 连接到前一个元素
                self.gst_core.link_elements(previous_element, element_name)
                previous_element = element_name
            
            # 添加输出格式转换
            self.gst_core.add_element("videoconvert", "video-convert-out")
            self.gst_core.link_elements(previous_element, "video-convert-out")
            
            # 添加输出接收器
            if output_sink == "display":
                self.gst_core.add_element("autovideosink", "video-sink")
            elif output_sink.endswith(('.mp4', '.avi', '.mkv')):
                # 文件输出
                self.gst_core.add_element("x264enc", "video-encoder")
                
                # 选择复用器
                if output_sink.endswith('.mp4'):
                    muxer = "mp4mux"
                elif output_sink.endswith('.avi'):
                    muxer = "avimux"
                else:
                    muxer = "matroskamux"
                
                self.gst_core.add_element(muxer, "muxer")
                self.gst_core.add_element("filesink", "file-sink")
                self.gst_core.set_element_property("file-sink", "location", output_sink)
                
                # 连接编码链
                self.gst_core.link_elements("video-convert-out", "video-encoder")
                self.gst_core.link_elements("video-encoder", "muxer")
                self.gst_core.link_elements("muxer", "file-sink")
            else:
                # 默认显示输出
                self.gst_core.add_element("autovideosink", "video-sink")
                self.gst_core.link_elements("video-convert-out", "video-sink")
            
            # 连接输入链
            if input_source != "test" and not input_source.startswith('/dev/video'):
                # 文件输入需要连接解码器
                self.gst_core.link_elements("file-source", "decoder")
                # 注意：decodebin需要动态连接
            else:
                # 直接连接到转换器
                self.gst_core.link_elements("video-source", "video-convert-in")
            
            print(f"Video filter pipeline created with filters: {', '.join(filters)}")
            return True
        
        except Exception as e:
            print(f"Error creating filter pipeline: {e}")
            return False
    
    def apply_real_time_effects(self, effect_name: str, params: Dict[str, Any] = None) -> bool:
        """应用实时特效"""
        try:
            if not self.gst_core.create_pipeline(f"realtime-{effect_name}-pipeline"):
                return False
            
            params = params or {}
            
            # 创建实时特效管道
            self.gst_core.add_element("videotestsrc", "video-source")
            self.gst_core.add_element("videoconvert", "video-convert-in")
            
            # 添加特效元素
            effect_element = self.processing_elements['effects'].get(effect_name)
            if not effect_element:
                print(f"Unknown effect: {effect_name}")
                return False
            
            self.gst_core.add_element(effect_element, "effect")
            
            # 设置特效参数
            for prop, value in params.items():
                self.gst_core.set_element_property("effect", prop, value)
            
            self.gst_core.add_element("videoconvert", "video-convert-out")
            self.gst_core.add_element("autovideosink", "video-sink")
            
            # 连接管道
            self.gst_core.link_elements("video-source", "video-convert-in")
            self.gst_core.link_elements("video-convert-in", "effect")
            self.gst_core.link_elements("effect", "video-convert-out")
            self.gst_core.link_elements("video-convert-out", "video-sink")
            
            print(f"Real-time effect '{effect_name}' pipeline created")
            return True
        
        except Exception as e:
            print(f"Error applying real-time effect: {e}")
            return False
    
    def start_processing(self) -> bool:
        """开始处理"""
        return self.gst_core.start_pipeline()
    
    def stop_processing(self) -> bool:
        """停止处理"""
        return self.gst_core.stop_pipeline()
    
    def cleanup(self):
        """清理资源"""
        self.gst_core.cleanup()
    
    def get_available_filters(self) -> Dict[str, List[str]]:
        """获取可用滤镜列表"""
        return {
            category: list(elements.keys()) 
            for category, elements in self.processing_elements.items()
        }

class AudioProcessor:
    """音频处理器"""
    
    def __init__(self):
        self.gst_core = GStreamerCore()
        self.audio_effects = {
            'equalizer': 'equalizer-10bands',
            'reverb': 'freeverb',
            'echo': 'audioecho',
            'pitch': 'pitch',
            'speed': 'speed',
            'volume': 'volume',
            'compressor': 'audiocompressor',
            'limiter': 'audiolimiter'
        }
    
    def create_audio_pipeline(self, input_source: str, output_sink: str, 
                             effects: List[str], effect_params: Dict[str, Dict] = None) -> bool:
        """创建音频处理管道"""
        try:
            if not self.gst_core.create_pipeline("audio-processing-pipeline"):
                return False
            
            effect_params = effect_params or {}
            
            # 添加输入源
            if input_source == "test":
                self.gst_core.add_element("audiotestsrc", "audio-source")
            elif input_source.startswith("alsa:"):
                self.gst_core.add_element("alsasrc", "audio-source")
                device = input_source.replace("alsa:", "")
                if device:
                    self.gst_core.set_element_property("audio-source", "device", device)
            else:
                # 文件输入
                self.gst_core.add_element("filesrc", "file-source")
                self.gst_core.set_element_property("file-source", "location", input_source)
                self.gst_core.add_element("decodebin", "decoder")
            
            # 添加格式转换
            self.gst_core.add_element("audioconvert", "audio-convert-in")
            self.gst_core.add_element("audioresample", "audio-resample-in")
            
            # 添加音效链
            previous_element = "audio-resample-in"
            for i, effect_name in enumerate(effects):
                effect_element = self.audio_effects.get(effect_name)
                if not effect_element:
                    print(f"Unknown audio effect: {effect_name}")
                    continue
                
                element_name = f"effect-{i}-{effect_name}"
                self.gst_core.add_element(effect_element, element_name)
                
                # 设置效果参数
                if effect_name in effect_params:
                    for prop, value in effect_params[effect_name].items():
                        self.gst_core.set_element_property(element_name, prop, value)
                
                # 连接到前一个元素
                self.gst_core.link_elements(previous_element, element_name)
                previous_element = element_name
            
            # 添加输出格式转换
            self.gst_core.add_element("audioconvert", "audio-convert-out")
            self.gst_core.add_element("audioresample", "audio-resample-out")
            self.gst_core.link_elements(previous_element, "audio-convert-out")
            self.gst_core.link_elements("audio-convert-out", "audio-resample-out")
            
            # 添加输出接收器
            if output_sink == "speaker":
                self.gst_core.add_element("autoaudiosink", "audio-sink")
                self.gst_core.link_elements("audio-resample-out", "audio-sink")
            elif output_sink.endswith(('.mp3', '.wav', '.ogg')):
                # 文件输出
                if output_sink.endswith('.mp3'):
                    encoder = "lamemp3enc"
                elif output_sink.endswith('.wav'):
                    encoder = "wavenc"
                else:
                    encoder = "vorbisenc"
                
                self.gst_core.add_element(encoder, "audio-encoder")
                self.gst_core.add_element("filesink", "file-sink")
                self.gst_core.set_element_property("file-sink", "location", output_sink)
                
                self.gst_core.link_elements("audio-resample-out", "audio-encoder")
                self.gst_core.link_elements("audio-encoder", "file-sink")
            
            # 连接输入链
            if input_source != "test" and not input_source.startswith("alsa:"):
                self.gst_core.link_elements("file-source", "decoder")
                # 注意：decodebin需要动态连接
            else:
                self.gst_core.link_elements("audio-source", "audio-convert-in")
                self.gst_core.link_elements("audio-convert-in", "audio-resample-in")
            
            print(f"Audio processing pipeline created with effects: {', '.join(effects)}")
            return True
        
        except Exception as e:
            print(f"Error creating audio pipeline: {e}")
            return False
    
    def start_processing(self) -> bool:
        """开始处理"""
        return self.gst_core.start_pipeline()
    
    def stop_processing(self) -> bool:
        """停止处理"""
        return self.gst_core.stop_pipeline()
    
    def cleanup(self):
        """清理资源"""
        self.gst_core.cleanup()
    
    def get_available_effects(self) -> List[str]:
        """获取可用音效列表"""
        return list(self.audio_effects.keys())

# 媒体处理演示
def demo_media_processing():
    print("Media Processing Demo")
    print("=====================")
    
    # 1. 视频滤镜处理
    print("\n1. Video Filter Processing")
    
    video_processor = VideoProcessor()
    
    # 显示可用滤镜
    available_filters = video_processor.get_available_filters()
    print("Available filters:")
    for category, filters in available_filters.items():
        print(f"  {category}: {', '.join(filters)}")
    
    # 创建滤镜管道
    filters = ['blur', 'edge']
    filter_params = {
        'blur': {'sigma': 2.0},
        'edge': {'threshold': 0.3}
    }
    
    if video_processor.create_filter_pipeline("test", "display", filters, filter_params):
        print(f"\nCreated video filter pipeline with: {', '.join(filters)}")
        
        # 短时间运行
        if video_processor.start_processing():
            print("Running video filters for 3 seconds...")
            time.sleep(3)
            video_processor.stop_processing()
    
    video_processor.cleanup()
    
    # 2. 实时视频特效
    print("\n2. Real-time Video Effects")
    
    effect_processor = VideoProcessor()
    
    # 应用aging特效
    aging_params = {
        'scratch-lines': 7,
        'color-aging': True,
        'pits': True
    }
    
    if effect_processor.apply_real_time_effects('aging', aging_params):
        print("Applied aging effect")
        
        if effect_processor.start_processing():
            print("Running aging effect for 3 seconds...")
            time.sleep(3)
            effect_processor.stop_processing()
    
    effect_processor.cleanup()
    
    # 3. 音频处理
    print("\n3. Audio Processing")
    
    audio_processor = AudioProcessor()
    
    # 显示可用音效
    available_effects = audio_processor.get_available_effects()
    print(f"Available audio effects: {', '.join(available_effects)}")
    
    # 创建音频处理管道
    effects = ['equalizer', 'reverb', 'volume']
    effect_params = {
        'equalizer': {
            'band0': -3.0,  # 60Hz
            'band1': 2.0,   # 170Hz
            'band2': 1.0    # 310Hz
        },
        'reverb': {
            'room-size': 0.8,
            'damping': 0.5,
            'wet-level': 0.3
        },
        'volume': {
            'volume': 1.5
        }
    }
    
    if audio_processor.create_audio_pipeline("test", "speaker", effects, effect_params):
        print(f"\nCreated audio processing pipeline with: {', '.join(effects)}")
        
        # 短时间运行
        if audio_processor.start_processing():
            print("Running audio effects for 3 seconds...")
            time.sleep(3)
            audio_processor.stop_processing()
    
    audio_processor.cleanup()
    
    print("\nMedia processing demo completed")

# 运行媒体处理演示
if __name__ == "__main__":
    demo_media_processing()

4.2 流媒体服务器

GStreamer可以构建强大的流媒体服务器：

class StreamingServer:
    """流媒体服务器"""
    
    def __init__(self, server_name: str = "gstreamer-server"):
        self.server_name = server_name
        self.active_streams = {}
        self.server_config = {
            'rtmp_port': 1935,
            'http_port': 8080,
            'udp_port': 5000,
            'max_clients': 100
        }
    
    def create_rtmp_server(self, rtmp_port: int = 1935) -> bool:
        """创建RTMP服务器"""
        try:
            # 创建RTMP接收管道
            rtmp_core = GStreamerCore()
            if not rtmp_core.create_pipeline("rtmp-server-pipeline"):
                return False
            
            # RTMP源
            rtmp_core.add_element("rtmpsrc", "rtmp-source")
            rtmp_core.set_element_property("rtmp-source", "location", f"rtmp://localhost:{rtmp_port}/live/stream")
            
            # 解复用
            rtmp_core.add_element("flvdemux", "demuxer")
            
            # 视频分支
            rtmp_core.add_element("queue", "video-queue")
            rtmp_core.add_element("h264parse", "video-parser")
            rtmp_core.add_element("avdec_h264", "video-decoder")
            rtmp_core.add_element("videoconvert", "video-convert")
            
            # 音频分支
            rtmp_core.add_element("queue", "audio-queue")
            rtmp_core.add_element("aacparse", "audio-parser")
            rtmp_core.add_element("faad", "audio-decoder")
            rtmp_core.add_element("audioconvert", "audio-convert")
            
            # 输出分支 - HLS
            rtmp_core.add_element("x264enc", "hls-video-encoder")
            rtmp_core.add_element("faac", "hls-audio-encoder")
            rtmp_core.add_element("flvmux", "hls-muxer")
            rtmp_core.add_element("hlssink", "hls-sink")
            
            # 设置HLS参数
            rtmp_core.set_element_property("hls-sink", "location", "stream_%05d.ts")
            rtmp_core.set_element_property("hls-sink", "playlist-location", "stream.m3u8")
            rtmp_core.set_element_property("hls-sink", "target-duration", 10)
            
            # 连接管道
            rtmp_core.link_elements("rtmp-source", "demuxer")
            
            # 注意：flvdemux有动态pad，需要特殊处理
            print(f"RTMP server pipeline created on port {rtmp_port}")
            
            self.active_streams['rtmp'] = rtmp_core
            return True
        
        except Exception as e:
            print(f"Error creating RTMP server: {e}")
            return False
    
    def create_udp_multicast_server(self, multicast_ip: str = "224.1.1.1", port: int = 5000) -> bool:
        """创建UDP组播服务器"""
        try:
            udp_core = GStreamerCore()
            if not udp_core.create_pipeline("udp-multicast-server"):
                return False
            
            # 视频源（可以是文件或摄像头）
            udp_core.add_element("videotestsrc", "video-source")
            udp_core.add_element("videoconvert", "video-convert")
            udp_core.add_element("x264enc", "video-encoder")
            
            # 音频源
            udp_core.add_element("audiotestsrc", "audio-source")
            udp_core.add_element("audioconvert", "audio-convert")
            udp_core.add_element("faac", "audio-encoder")
            
            # 复用
            udp_core.add_element("mpegtsmux", "muxer")
            
            # UDP输出
            udp_core.add_element("udpsink", "udp-sink")
            udp_core.set_element_property("udp-sink", "host", multicast_ip)
            udp_core.set_element_property("udp-sink", "port", port)
            udp_core.set_element_property("udp-sink", "auto-multicast", True)
            
            # 连接管道
            udp_core.link_elements("video-source", "video-convert")
            udp_core.link_elements("video-convert", "video-encoder")
            udp_core.link_elements("video-encoder", "muxer")
            
            udp_core.link_elements("audio-source", "audio-convert")
            udp_core.link_elements("audio-convert", "audio-encoder")
            udp_core.link_elements("audio-encoder", "muxer")
            
            udp_core.link_elements("muxer", "udp-sink")
            
            print(f"UDP multicast server created: {multicast_ip}:{port}")
            
            self.active_streams['udp_multicast'] = udp_core
            return True
        
        except Exception as e:
            print(f"Error creating UDP multicast server: {e}")
            return False
    
    def create_http_streaming_server(self, http_port: int = 8080) -> bool:
        """创建HTTP流媒体服务器"""
        try:
            http_core = GStreamerCore()
            if not http_core.create_pipeline("http-streaming-server"):
                return False
            
            # 视频源
            http_core.add_element("videotestsrc", "video-source")
            http_core.add_element("videoconvert", "video-convert")
            http_core.add_element("x264enc", "video-encoder")
            
            # 设置编码参数
            http_core.set_element_property("video-encoder", "tune", "zerolatency")
            http_core.set_element_property("video-encoder", "bitrate", 1000)
            
            # HTTP输出
            http_core.add_element("mpegtsmux", "muxer")
            http_core.add_element("tcpserversink", "http-sink")
            http_core.set_element_property("http-sink", "host", "0.0.0.0")
            http_core.set_element_property("http-sink", "port", http_port)
            
            # 连接管道
            http_core.link_elements("video-source", "video-convert")
            http_core.link_elements("video-convert", "video-encoder")
            http_core.link_elements("video-encoder", "muxer")
            http_core.link_elements("muxer", "http-sink")
            
            print(f"HTTP streaming server created on port {http_port}")
            
            self.active_streams['http'] = http_core
            return True
        
        except Exception as e:
            print(f"Error creating HTTP streaming server: {e}")
            return False
    
    def start_all_servers(self) -> bool:
        """启动所有服务器"""
        success = True
        for stream_name, stream_core in self.active_streams.items():
            if not stream_core.start_pipeline():
                print(f"Failed to start {stream_name} server")
                success = False
            else:
                print(f"{stream_name} server started successfully")
        return success
    
    def stop_all_servers(self) -> bool:
        """停止所有服务器"""
        success = True
        for stream_name, stream_core in self.active_streams.items():
            if not stream_core.stop_pipeline():
                print(f"Failed to stop {stream_name} server")
                success = False
            else:
                print(f"{stream_name} server stopped")
        return success
    
    def get_server_status(self) -> Dict[str, Dict]:
        """获取服务器状态"""
        status = {}
        for stream_name, stream_core in self.active_streams.items():
            pipeline_info = stream_core.get_pipeline_info()
            status[stream_name] = {
                'state': pipeline_info.get('state', 'unknown'),
                'elements': pipeline_info.get('elements_count', 0),
                'running': stream_core.is_playing
            }
        return status
    
    def cleanup(self):
        """清理所有资源"""
        for stream_core in self.active_streams.values():
            stream_core.cleanup()
        self.active_streams.clear()
        print(f"Streaming server '{self.server_name}' cleaned up")

# 流媒体服务器演示
def demo_streaming_server():
    print("Streaming Server Demo")
    print("=====================")
    
    server = StreamingServer("demo-streaming-server")
    
    # 1. 创建不同类型的流媒体服务器
    print("\n1. Creating Streaming Servers")
    
    # UDP组播服务器
    if server.create_udp_multicast_server("224.1.1.1", 5000):
        print("UDP multicast server created")
    
    # HTTP流媒体服务器
    if server.create_http_streaming_server(8080):
        print("HTTP streaming server created")
    
    # 2. 显示服务器状态
    print("\n2. Server Status (before start)")
    status = server.get_server_status()
    for server_name, server_info in status.items():
        print(f"  {server_name}: {server_info['state']} ({server_info['elements']} elements)")
    
    # 3. 启动服务器
    print("\n3. Starting Servers")
    if server.start_all_servers():
        print("All servers started successfully")
        
        # 运行一段时间
        print("\nServers running for 5 seconds...")
        time.sleep(5)
        
        # 显示运行状态
        print("\n4. Server Status (running)")
        status = server.get_server_status()
        for server_name, server_info in status.items():
            print(f"  {server_name}: {server_info['state']} (running: {server_info['running']})")
    
    # 4. 停止服务器
    print("\n5. Stopping Servers")
    server.stop_all_servers()
    
    # 清理资源
    server.cleanup()
    
    print("\nStreaming server demo completed")

# 运行流媒体服务器演示
if __name__ == "__main__":
    demo_streaming_server()

5. 总结与实际应用

5.1 关键技术要点

GStreamer作为强大的多媒体框架，具有以下核心优势：

1. 模块化架构：基于插件的设计使得功能扩展灵活
2. 跨平台支持：支持Linux、Windows、macOS、Android等平台
3. 丰富的插件生态：涵盖编解码、滤镜、网络传输等各个方面
4. 实时处理能力：支持低延迟的实时音视频处理
5. 管道式设计：直观的数据流处理模型

5.2 技术发展趋势

1. 硬件加速集成：更好地利用GPU、专用编解码芯片
2. AI增强处理：集成机器学习算法进行智能音视频处理
3. 云原生支持：适应容器化和微服务架构
4. WebRTC集成：更好地支持Web实时通信
5. 新编解码标准：支持AV1、VVC等新一代编解码标准

5.3 实际应用建议

class GStreamerApplicationGuide:
    """GStreamer应用指南"""
    
    def __init__(self):
        self.application_scenarios = {
            'media_player': {
                'description': '多媒体播放器开发',
                'key_elements': ['filesrc', 'decodebin', 'autovideosink', 'autoaudiosink'],
                'considerations': ['格式支持', '硬件解码', '字幕支持', '播放控制']
            },
            'live_streaming': {
                'description': '直播流媒体应用',
                'key_elements': ['v4l2src', 'x264enc', 'flvmux', 'rtmpsink'],
                'considerations': ['延迟控制', '码率自适应', '网络稳定性', '多路推流']
            },
            'video_conferencing': {
                'description': '视频会议系统',
                'key_elements': ['webrtcbin', 'dtlssrtpenc', 'dtlssrtpdec', 'opusenc'],
                'considerations': ['回声消除', '噪声抑制', 'NAT穿透', '多方通话']
            },
            'surveillance': {
                'description': '视频监控系统',
                'key_elements': ['rtspsrc', 'h264parse', 'mp4mux', 'splitmuxsink'],
                'considerations': ['存储管理', '运动检测', '多路录制', '远程访问']
            },
            'transcoding': {
                'description': '视频转码服务',
                'key_elements': ['filesrc', 'decodebin', 'videoconvert', 'x264enc'],
                'considerations': ['批量处理', '质量控制', '性能优化', '格式支持']
            }
        }
    
    def get_application_guide(self, scenario: str) -> Dict[str, Any]:
        """获取应用场景指南"""
        return self.application_scenarios.get(scenario, {})
    
    def recommend_pipeline_architecture(self, scenario: str, requirements: Dict[str, Any]) -> Dict[str, Any]:
        """推荐管道架构"""
        guide = self.get_application_guide(scenario)
        if not guide:
            return {}
        
        recommendations = {
            'scenario': scenario,
            'description': guide['description'],
            'recommended_elements': guide['key_elements'],
            'key_considerations': guide['considerations'],
            'pipeline_suggestions': []
        }
        
        # 根据需求提供具体建议
        if scenario == 'media_player':
            if requirements.get('hardware_decode', False):
                recommendations['pipeline_suggestions'].append(
                    "Use hardware decoders like vaapidecodebin for better performance"
                )
            if requirements.get('subtitle_support', False):
                recommendations['pipeline_suggestions'].append(
                    "Add textoverlay element for subtitle rendering"
                )
        
        elif scenario == 'live_streaming':
            if requirements.get('low_latency', False):
                recommendations['pipeline_suggestions'].append(
                    "Use tune=zerolatency for x264enc and minimize buffer sizes"
                )
            if requirements.get('adaptive_bitrate', False):
                recommendations['pipeline_suggestions'].append(
                    "Implement bitrate control based on network conditions"
                )
        
        return recommendations
    
    def get_performance_optimization_tips(self) -> List[str]:
        """获取性能优化建议"""
        return [
            "使用硬件加速元素（如vaapi、nvenc）减少CPU负载",
            "合理设置缓冲区大小，平衡延迟和稳定性",
            "使用多线程处理，避免阻塞主管道",
            "选择合适的像素格式，减少不必要的转换",
            "监控管道性能，及时发现瓶颈",
            "使用Pad探针进行数据分析和调试",
            "优化编码参数，在质量和性能间找到平衡",
            "合理使用队列元素，避免数据流阻塞"
        ]
    
    def get_debugging_strategies(self) -> List[str]:
        """获取调试策略"""
        return [
            "使用GST_DEBUG环境变量启用详细日志",
            "通过gst-inspect工具查看元素属性和Caps",
            "使用gst-launch命令行工具快速测试管道",
            "添加Pad探针监控数据流",
            "检查元素状态变化和错误消息",
            "使用可视化工具（如gst-shark）分析管道性能",
            "逐步构建管道，定位问题元素",
            "验证Caps兼容性，确保元素正确连接"
        ]

# 应用指南演示
def demo_application_guide():
    print("GStreamer Application Guide Demo")
    print("=================================")
    
    guide = GStreamerApplicationGuide()
    
    # 1. 显示应用场景
    print("\n1. Available Application Scenarios")
    for scenario, info in guide.application_scenarios.items():
        print(f"  {scenario}: {info['description']}")
    
    # 2. 获取特定场景的指南
    print("\n2. Live Streaming Application Guide")
    
    streaming_requirements = {
        'low_latency': True,
        'adaptive_bitrate': True,
        'multi_resolution': False
    }
    
    recommendations = guide.recommend_pipeline_architecture('live_streaming', streaming_requirements)
    
    print(f"Scenario: {recommendations['description']}")
    print(f"Key elements: {', '.join(recommendations['recommended_elements'])}")
    print(f"Considerations: {', '.join(recommendations['key_considerations'])}")
    print("Pipeline suggestions:")
    for suggestion in recommendations['pipeline_suggestions']:
        print(f"  - {suggestion}")
    
    # 3. 性能优化建议
    print("\n3. Performance Optimization Tips")
    tips = guide.get_performance_optimization_tips()
    for i, tip in enumerate(tips, 1):
        print(f"  {i}. {tip}")
    
    # 4. 调试策略
    print("\n4. Debugging Strategies")
    strategies = guide.get_debugging_strategies()
    for i, strategy in enumerate(strategies, 1):
        print(f"  {i}. {strategy}")
    
    print("\nApplication guide demo completed")

# 运行应用指南演示
if __name__ == "__main__":
    demo_application_guide()

通过本文的深入探讨，我们全面了解了GStreamer多媒体框架的核心架构、关键组件和实际应用。GStreamer以其强大的功能、灵活的设计和丰富的生态系统，为多媒体应用开发提供了坚实的基础。

无论是构建简单的媒体播放器，还是开发复杂的流媒体服务，GStreamer都能提供相应的解决方案。随着技术的不断发展，GStreamer也在持续演进，集成更多新技术和标准，为开发者创造更多可能性。

掌握GStreamer不仅能够帮助我们更好地理解多媒体处理的原理，还能为实际项目开发提供强有力的技术支撑。希望本文能够为读者在多媒体开发道路上提供有价值的参考和指导。

版权所有，如有侵权请联系我