本文详解如何在 python 中高效、低丢帧地处理来自 eufy 安全 websocket 服务的 h.264 实时视频流，重点解决帧不完整、解码器重复创建及 p/b 帧丢失等常见问题。
H.264 是一种典型的基于帧间预测（inter-frame）的压缩编码格式，其视频流由 I 帧（关键帧）、P 帧（前向预测）和 B 帧（双向预测）组成。Eufy WebSocket 服务推送的并非独立 JPEG 或完整 GOP（Group of Pictures），而是连续的、可能被截断的 NAL 单元（Network Abstraction Layer units）字节流——这意味着单次 on_message 接收的数据往往只是某个帧的一部分，或跨多个帧的碎片。原始代码中 is_h264_complete() 的逻辑存在根本性误解：它试图用 SPS/PPS（序列/图像参数集）的存在来判断“一帧是否完整”，但 H.264 的帧完整性不能靠静态字节扫描判定；SPS/PPS 通常只在流起始或关键帧前发送一次，而后续 P/B 帧完全依赖解码器内部状态重建。
更严重的是，原代码在每次收到消息时都重新创建 av.CodecContext 并调用 codec.decode(packet)，这不仅带来巨大开销，更导致解码器无法维持上下文（如参考帧队列），从而彻底丢弃所有非 I 帧（即 P/B 帧），最终仅能显示稀疏、卡顿的关键帧画面。
✅ 正确做法是：将解码器生命周期与 WebSocket 连接对齐，复用 CodecContext，并以流式方式持续喂入原始字节包。以下是优化后的核心实现：
import websocket
import json
import av
import cv2
import numpy as np
# 全局解码器（复用，避免重复初始化）
codec = None
frame_buffer = bytearray() # 累积未完成的NALU数据
def on_message(ws, message):
global codec, frame_buffer
data = json.loads(message)
if data.get("type") == "event" and data.get("event", {}).get("event") == "livestream video data":
# 获取原始字节数据（base64 解码后为 bytes）
raw_bytes = bytes(data["event"]["buffer"]["data"])
frame_buffer.extend(raw_bytes)
# 尝试从累积缓冲区中提取完整的 NAL 单元（以 0x00000001 或 0x000001 开头）
# 注意：Eufy 流通常使用 4-byte start code
start_codes = [b'\x00\x00\x00\x01', b'\x00\x00\x01']
packets = []
offset = 0
while offset < len(frame_buffer):
found = False
for sc in start_codes:
pos = frame_buffer.find(sc, offset)
if pos != -1:
if offset > 0:
# 提取上一个 start code 到当前 start code 之间的 NALU
packets.append(frame_buffer[offset:pos])
offset = pos + len(sc)
found = True
break
if not found:
break
# 清理已提取部分，保留末尾不完整 NALU
if packets:
# 移除已处理的完整 NALUs（含 start code）
last_end = 0
for p in packets:
# 找到该 packet 在 buffer 中的实际起始位置（含 start code）
for sc in start_codes:
idx = frame_buffer.find(sc, last_end)
if idx != -1:
last_end = idx + len(sc) + len(p)
break
frame_buffer = frame_buffer[last_end:]
# 初始化解码器（首次遇到 SPS/PPS 时）
if codec is None:
# 检查是否有 SPS (NALU type 7) 或 PPS (type 8)
for pkt in packets:
if len(pkt) >= 5:
nalu_type = pkt[0] & 0x1F
if nalu_type == 7 or nalu_type == 8:
# 创建解码器（仅一次）
codec = av.CodecContext.create('h264', 'r')
break
# 解码所有提取出的 NALU packets
if codec is not None:
for pkt_bytes in packets:
try:
packet = av.Packet(pkt_bytes)
frames = codec.decode(packet)
for frame in frames:
img = frame.to_ndarray(format='bgr24')
# 可选：叠加时间戳或帧序号增强调试
cv2.putText(img, f"FPS: {int(codec.framerate)}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
cv2.imshow('Eufy Live Stream', img)
if cv2.waitKey(1) & 0xFF == ord('q'):
ws.close()
return
except Exception as e:
print(f"[Decode Error] {e}")
continue
def on_open(ws):
print("✅ WebSocket connected")
ws.send(json.dumps({"messageId": "start_listening", "command": "start_listening"}))
ws.send(json.dumps({"command": "set_api_schema", "schemaVersion": 20}))
ws.send(json.dumps({
"messageId": "start_livestream",
"command": "device.start_livestream",
"serialNumber": "T8410P4223334EBE" # 替换为你的设备序列号
}))
def on_close(ws):
global codec
print("? Connection closed")
if codec:
codec.close()
codec = None
cv2.destroyAllWindows()
def on_error(ws, error):
print(f"⚠️ WebSocket error: {error}")
if __name__ == "__main__":
websocket.enableTrace(False)
ws = websocket.WebSocketApp(
"ws://localhost:3000",
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
ws.run_forever()? 关键改进说明：
立即学习“Python免费学习笔记（深入）”；
- NALU 边界识别：不再依赖错误的 is_h264_complete()，而是按标准 H.264 起始码（0x00000001 或 0x000001）切分原始字节流，确保每个 av.Packet 对应一个语义完整的 NAL 单元。
- 解码器单例化：av.CodecContext 在首次检测到 SPS（type 7）或 PPS（type 8）时创建，并在整个连接生命周期内复用，保障 P/B 帧可被正确参考解码。
- 流式累积与清理：frame_buffer 持续接收 WebSocket 数据，每次 on_message 都尝试提取已完成的 NALU，未完成部分保留在缓冲区，避免帧碎片丢失。
- 异常鲁棒性：对单个 packet 解码失败做静默跳过，防止因网络抖动或数据损坏导致整个流中断。
? 额外建议：
- 若仍出现卡顿，可增加 cv2.waitKey(1) 的延时（如 waitKey(5)）或启用 OpenCV 的硬件加速后端（如 cv2.CAP_FFMPEG）；
- 生产环境建议使用 threading.Lock 保护 frame_buffer，避免多线程竞争（当前单线程 WebSocket 回调暂无需）；
- 如需音频同步，Eufy 流通常分离传输 AAC 音频，需另建解码通道并使用 PTS/DTS 对齐。
通过以上重构，你将获得流畅、低延迟、全帧率（I+P+B）的 Eufy 实时视频流渲染能力。
