175 lines
6.2 KiB
Python
175 lines
6.2 KiB
Python
# 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
|