's'

stream_stats.py

@@ -0,0 +1,174 @@
+# stream_stats.py
+# 推流状态监控：解析 FFmpeg 输出，统计帧率、码率、网络等，便于调试
+"""
+推流状态监控模块
+解析 FFmpeg stderr 输出，提取：帧率、码率、编码速度、源延迟事件
+可选：Linux 网络上下行速度
+"""
+import re
+import time
+import threading
+import platform
+
+
+class StreamStats:
+    """推流状态统计，线程安全"""
+
+    def __init__(self, interval_sec=10):
+        self.interval_sec = interval_sec
+        self._lock = threading.Lock()
+        self.frame = 0
+        self.fps = 0.0
+        self.bitrate_kbps = 0.0
+        self.speed = 0.0
+        self.encode_time = ""
+        self.size_kb = 0
+        self.quality_q = 0.0
+        self.lag_events = 0  # "Resumed reading" 次数，表示源端抖动
+        self.last_update_ts = 0.0
+        self.start_ts = time.time()
+        self._stop = threading.Event()
+        self._printer_thread = None
+        # 网络统计 (Linux)
+        self._net_last = None
+        self._net_last_ts = 0
+        self.rx_mbps = 0.0
+        self.tx_mbps = 0.0
+
+    def parse_ffmpeg_progress(self, line: str) -> bool:
+        """解析 FFmpeg 进度行，返回是否成功解析"""
+        # frame=  12345 fps= 30 q=28.0 size=   12345kB time=00:06:51.50 bitrate=2460.0kbits/s speed=1.00x
+        m = re.search(r'frame=\s*(\d+)', line, re.I)
+        if not m:
+            return False
+        with self._lock:
+            try:
+                self.frame = int(m.group(1))
+            except (ValueError, IndexError):
+                pass
+            m2 = re.search(r'fps=\s*([\d.]+)', line, re.I)
+            if m2:
+                self.fps = float(m2.group(1))
+            m3 = re.search(r'bitrate=\s*([\d.]+)\s*kbits/s', line, re.I)
+            if m3:
+                self.bitrate_kbps = float(m3.group(1))
+            m4 = re.search(r'speed=\s*([\d.]+)x', line, re.I)
+            if m4:
+                self.speed = float(m4.group(1))
+            m5 = re.search(r'time=([^\s]+)', line, re.I)
+            if m5:
+                self.encode_time = m5.group(1).strip()
+            m6 = re.search(r'size=\s*(\d+)\s*kB', line, re.I)
+            if m6:
+                self.size_kb = int(m6.group(1))
+            m7 = re.search(r'q=([\d.]+)', line, re.I)
+            if m7:
+                self.quality_q = float(m7.group(1))
+            self.last_update_ts = time.time()
+        return True
+
+    def parse_lag_event(self, line: str) -> bool:
+        """检测源端延迟事件 Resumed reading at pts"""
+        if "resumed reading at pts" in line.lower():
+            with self._lock:
+                self.lag_events += 1
+            return True
+        return False
+
+    def _read_net_bytes(self):
+        """读取 /proc/net/dev 总字节数 (Linux)"""
+        try:
+            with open("/proc/net/dev", "r") as f:
+                content = f.read()
+        except Exception:
+            return None, None
+        rx_total = tx_total = 0
+        for line in content.split("\n")[2:]:  # 跳过表头
+            line = line.strip()
+            if not line or ":" not in line:
+                continue
+            parts = line.split()
+            if len(parts) >= 10:
+                try:
+                    rx_total += int(parts[1])
+                    tx_total += int(parts[9])
+                except (ValueError, IndexError):
+                    pass
+        return rx_total, tx_total
+
+    def update_network(self):
+        """更新网络速度 (Linux)"""
+        if platform.system() != "Linux":
+            return
+        rx, tx = self._read_net_bytes()
+        if rx is None:
+            return
+        now = time.time()
+        with self._lock:
+            if self._net_last is not None:
+                dt = now - self._net_last_ts
+                if dt > 0.5:  # 至少 0.5 秒间隔
+                    drx = (rx - self._net_last[0]) / (1024 * 1024) / dt
+                    dtx = (tx - self._net_last[1]) / (1024 * 1024) / dt
+                    self.rx_mbps = round(drx, 2)
+                    self.tx_mbps = round(dtx, 2)
+            self._net_last = (rx, tx)
+            self._net_last_ts = now
+
+    def format_summary(self) -> str:
+        """生成状态摘要字符串"""
+        self.update_network()
+        with self._lock:
+            elapsed = time.time() - self.start_ts
+            parts = [
+                f"推流时长 {int(elapsed)}s",
+                f"帧率 {self.fps:.1f} fps",
+                f"已推帧 {self.frame}",
+                f"码率 {self.bitrate_kbps:.0f} kbps",
+                f"编码速度 {self.speed:.2f}x",
+            ]
+            if self.lag_events > 0:
+                parts.append(f"源抖动 {self.lag_events} 次")
+            if platform.system() == "Linux":
+                parts.append(f"下行 {self.rx_mbps:.2f} MB/s 上行 {self.tx_mbps:.2f} MB/s")
+        return " | ".join(parts)
+
+    def start_periodic_printer(self):
+        """启动周期性打印线程"""
+
+        def _run():
+            while not self._stop.is_set():
+                self._stop.wait(self.interval_sec)
+                if self._stop.is_set():
+                    break
+                if self.last_update_ts > 0 and (time.time() - self.last_update_ts) < 60:
+                    print(f"[推流状态] {self.format_summary()}")
+
+        self._printer_thread = threading.Thread(target=_run, daemon=True)
+        self._printer_thread.start()
+
+    def stop(self):
+        """停止周期性打印"""
+        self._stop.set()
+
+
+def parse_and_maybe_print(line: str, stats: StreamStats, prefix: str = "[FFmpeg]", verbose=True) -> bool:
+    """
+    解析一行 FFmpeg 输出，更新 stats。
+    若为进度行则更新不打印；若为 lag 事件则更新并可选打印；其他行原样打印。
+    verbose: 是否打印非进度行
+    """
+    line_stripped = line.strip() if isinstance(line, str) else line.decode("utf-8", errors="replace").strip()
+    if not line_stripped:
+        return False
+    # 进度行：更新 stats，不重复打印（由周期摘要代替）
+    if stats.parse_ffmpeg_progress(line_stripped):
+        return True
+    # 源端延迟事件
+    if stats.parse_lag_event(line_stripped):
+        if verbose:
+            print(f"{prefix} {line_stripped}")
+        return True
+    if verbose:
+        print(f"{prefix} {line_stripped}")
+    return False