gst-launch-1.0 -v 
nvarguscamerasrc sensor-id=0 
    ! 'video/x-raw(memory:NVMM),width=1920,height=1080,framerate=30/1,format=NV12' 
    ! nvvidconv ! 'video/x-raw(memory:NVMM),format=I420' 
    ! nvh264enc preset=low-latency-hq rc-mode=cbr bitrate=8000 
        iframeinterval=30 insert-sps-pps=true maxperf-enable=1 
    ! h264parse config-interval=1 
    ! rtph264pay config-interval=1 pt=96 
    ! srtsink uri="srt://192.168.1.100:5000?mode=caller&latency=50&peerlatency=50&pbkeylen=0"

gst-launch-1.0 -v 
srtsrc uri="srt://:5000?mode=listener&latency=50" 
    ! application/x-rtp,media=video,encoding-name=H264,payload=96 
    ! rtph264depay ! h264parse 
    ! nvh264dec max-threads=0 output-corrupt=false 
    ! nv3dsink sync=false max-lateness=-1 qos=false

// 在 appsink 或自定义 probe 中获取 buffer PTS，转换为系统绝对时间
GstClock *net_clock = gst_net_client_clock_new("sender-clock", "192.168.1.100", 9999, 0);
GstClockTime sender_pts = gst_clock_get_time(net_clock);
// 将 sender_pts 作为 SEI 植入 H.264 码流（user_data_unregistered）

// 在 nvh264dec src pad probe 中解析 SEI 取出 sender_pts
GstClockTime recv_now = gst_clock_get_time(net_clock); // 同一时钟源
GstClockTime latency = recv_now - sender_pts;
g_print("E2E Latency: %" GST_TIME_FORMAT "n", GST_TIME_ARGS(latency));

┌─────────────────────────────────────────────────────────────┐
│                      接入层 (Ingress)                        │
│  SRT Listener / WebRTC (webrtcbin) / RTMP / GB28181 (SIP)   │
└─────────────────────┬───────────────────────────────────────┘
                      │ GstBuffer (NVMM/dmabuf) 零拷贝传递
┌─────────────────────▼───────────────────────────────────────┐
│                      核心调度层 (Core Scheduler)             │
│  • Stream Router (基于 Stream ID 动态挂载/卸载 Bin)          │
│  • Transcode Farm (nvh264enc/nvdec + 动态负载均衡)           │
│  • Record/TS/MP4 分片落盘 (multifilesink + 关键帧对齐)       │
│  • AI Pre-process (nvstreammux → nvinfer → nvdsosd)         │
└─────────────────────┬───────────────────────────────────────┘
                      │ 动态 Pad / Ghost Pad 暴露
┌─────────────────────▼───────────────────────────────────────┐
│                      分发层 (Egress)                         │
│  SRT Caller / WebRTC (mesh/sfu) / HLS (hlssink2) / GB28181  │
└─────────────────────────────────────────────────────────────┘

stateDiagram-v2
    [*] --> IDLE: 启动
    IDLE --> GATHERING: 创建 webrtcbin, 设置 stun/turn
    GATHERING --> LOCAL_DESC_SET: on-negotiation-needed -> create-offer -> set-local-desc
    LOCAL_DESC_SET --> SIGNALING: 发送 Offer (SDP + ICE Candidates)
    SIGNALING --> REMOTE_DESC_SET: 收到 Answer -> set-remote-desc
    REMOTE_DESC_SET --> CONNECTED: ICE State = CONNECTED/COMPLETED
    CONNECTED --> RE_NEGOTIATING: 码率自适应 / 重协商
    RE_NEGOTIATING --> LOCAL_DESC_SET
    CONNECTED --> [*]: EOS / Error

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

class WebRTCSender:
    def __init__(self, pipeline_desc, stun_server="stun://stun.l.google.com:19302"):
        self.pipe = Gst.parse_launch(pipeline_desc)
        self.webrtc = self.pipe.get_by_name("webrtcbin")
        self.webrtc.set_property("stun-server", stun_server)
        self.webrtc.set_property("bundle-policy", GstWebRTC.BundlePolicy.MAX_BUNDLE)
        
        # 关键信号连接
        self.webrtc.connect("on-negotiation-needed", self._on_negotiation_needed)
        self.webrtc.connect("on-ice-candidate", self._on_ice_candidate)
        self.webrtc.connect("on-data-channel", self._on_data_channel) # 可选数据通道
        
        # ICE 状态监控，超时自动重连
        self.webrtc.connect("notify::ice-connection-state", self._on_ice_state_change)
        
        self.bus = self.pipe.get_bus()
        self.bus.add_signal_watch()
        self.bus.connect("message::error", self._on_error)

    def _on_negotiation_needed(self, element):
        # 必须在主线程/GLib 主循环中执行
        promise = Gst.Promise.new_with_change_func(self._on_offer_created, element, None)
        element.emit("create-offer", None, promise)

    def _on_offer_created(self, promise, element, _):
        reply = promise.wait()
        offer = reply.get_structure().get_value("offer")
        element.emit("set-local-description", offer, None)
        # 通过信令通道发送 offer (JSON: {"type": "offer", "sdp": offer.sdp.as_text()})
        self.signal_send({"type": "offer", "sdp": offer.sdp.as_text()})

    def _on_ice_candidate(self, _, mline_index, candidate):
        # 信令发送 candidate: {"type": "candidate", "mline": mline_index, "candidate": candidate}
        self.signal_send({"type": "candidate", "mline": mline_index, "candidate": candidate})

    def handle_remote_answer(self, sdp_text):
        sdp = GstSdp.SDPMessage.new()
        GstSdp.sdp_message_parse_buffer(bytes(sdp_text, "utf-8"), sdp)
        answer = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.ANSWER, sdp)
        self.webrtc.emit("set-remote-description", answer, None)

    def handle_remote_candidate(self, mline_index, candidate_str):
        self.webrtc.emit("add-ice-candidate", mline_index, candidate_str)

    def _on_ice_state_change(self, _, param):
        state = self.webrtc.get_property("ice-connection-state")
        if state in (GstWebRTC.WebRTCICEConnectionState.FAILED, 
                     GstWebRTC.WebRTCICEConnectionState.DISCONNECTED):
            # 触发 ICE Restart 逻辑
            self.webrtc.emit("set-local-description", None, None) # 触发重新协商

// 自定义 Tracer: gst_tracer_record_log (buffer, "trace_id", "00-1234...-01")
static void
on_buffer_passed (GstTracer *tracer, GstPad *pad, GstBuffer *buffer, gpointer user_data)
{
    GstClockTime pts = GST_BUFFER_PTS(buffer);
    const gchar *trace_id = gst_buffer_get_meta(buffer, GST_TRACE_META_API_TYPE);
    if (!trace_id) {
        trace_id = generate_trace_id(); // W3C TraceContext 格式
        gst_buffer_add_trace_meta(buffer, trace_id);
    }
    // 上报至 OTel Collector
    otel_span_start("gstreamer.buffer.flow", trace_id, pts);
}

class HwEncoderFactory:
    @staticmethod
    def create_encoder(codec="h264", bitrate=8000, gop=30, preset="low-latency"):
        platform = detect_platform() # 读取 /proc/device-tree/compatible 或厂商 SDK 环境变量
        
        if platform == "rockchip": # RK3588 / RK3568
            return Gst.ElementFactory.make("mpph264enc" if codec=="h264" else "mpph265enc", None)
        elif platform == "hisi": # Hi3559A / Hi3519DV300
            return Gst.ElementFactory.make("vpuenc_h264" if codec=="h264" else "vpuenc_h265", None)
        elif platform == "tiandy": # 天邑 TYH264/265
            return Gst.ElementFactory.make("tyh264enc" if codec=="h264" else "tyh265enc", None)
        elif platform == "cambricon": # 寒武纪 MLU220/270
            return Gst.ElementFactory.make("cnenc_h264" if codec=="h264" else "cnenc_h265", None)
        else: # 兜底 x86/NVIDIA
            return Gst.ElementFactory.make("nvh264enc" if codec=="h264" else "nvh265enc", None)

    @staticmethod
    def configure_low_latency(encoder, codec, bitrate, gop):
        # 统一属性映射表，屏蔽厂商差异
        prop_map = {
            "rockchip":   {"rc_mode": "cbr", "bps": bitrate*1000, "gop": gop, "profile": "high"},
            "hisi":       {"rc_mode": "cbr", "bitrate": bitrate*1000, "gop": gop, "profile": 100}, # High Profile
            "tiandy":     {"rate_control": "cbr", "target_bitrate": bitrate*1000, "gop_size": gop},
            "cambricon":  {"rc_mode": "cbr", "bitrate": bitrate*1000, "gop_size": gop, "tune": "low_latency"},
            "nvidia":     {"preset": "low-latency-hq", "rc_mode": "cbr", "bitrate": bitrate*1000, "iframeinterval": gop}
        }
        for k, v in prop_map.get(detect_platform(), {}).items():
            encoder.set_property(k, v)

#!/bin/bash
# chaos_netem.sh: 注入 1% 丢包、50ms±10ms 抖动、1Mbps 限速
IFACE="eth0"
tc qdisc del dev $IFACE root 2>/dev/null
tc qdisc add dev $IFACE root handle 1: netem loss 1% delay 50ms 10ms distribution normal rate 1mbit
# 恢复: tc qdisc del dev $IFACE root

# 场景：模拟编码器内部错误、解码器输出损坏 Buffer、网络元素发送 EOS
gst-validate-1.0 --scenario="encoder_failure,decoder_corrupt,network_eos" 
    "your_pipeline_description" 
    --set-media-folder=/test/vectors 
    --output-folder=/test/results

# validate.scenario.yaml
actions:
  - type: "element-error"
    element-name: "nvh264enc"
    error-code: "GST_RESOURCE_ERROR"
    message: "Simulated NVENC session exhausted"
  - type: "expect"
    element-name: "pipeline"
    property: "current-state"
    value: "PLAYING"  # 期望管线自动重建 encoder 后恢复 PLAYING
    timeout: 5000     # 5秒内自愈

def _monitor_pipeline_health(self):
    while self.running:
        # 1. 检查关键元素状态
        enc = self.pipe.get_by_name("video_encoder")
        if enc.get_state(0)[1] != Gst.State.PLAYING:
            self.logger.warning("Encoder stalled, triggering rebuild...")
            self._rebuild_encoder_bin() # 热替换 Bin
        
        # 2. 检查队列积压 (背压信号)
        queue = self.pipe.get_by_name("net_queue")
        level_time = queue.get_property("current-level-time")
        if level_time > 500_000_000: # 500ms 积压
            self.logger.critical("Severe backpressure, dropping frames to recover")
            queue.set_property("max-size-time", 100_000_000) # 暴力降缓冲
            self._request_keyframe() # 强制 IDR 重同步
        
        time.sleep(1)