基于Eufy Security WebSocket Server的H264直播帧丢帧问题求助

问题背景 #

基于eufy-security-client构建的WebSocket Server，通过WebSocket接口获取设备直播流，用Python客户端调用device.start_livestream命令获取H264格式视频帧缓冲区，渲染时出现大量丢帧，推测是帧间压缩导致，尝试实现缓冲/包重组逻辑但未解决。

原代码：

import websocket
import json
import av
import cv2
buffer = bytearray()
def is_h264_complete(buffer):
    # Convert the buffer to bytes
    buffer_bytes = bytes(buffer)
    # Look for the start code in the buffer
    start_code = bytes([0, 0, 0, 1])
    positions = [i for i in range(len(buffer_bytes)) if buffer_bytes.startswith(start_code, i)]
    # Check for the presence of SPS and PPS
    has_sps = any(buffer_bytes[i+4] & 0x1F == 7 for i in positions)
    has_pps = any(buffer_bytes[i+4] & 0x1F == 8 for i in positions)
    return has_sps and has_pps
def on_message(ws, message):
    data = json.loads(message)
    message_type = data["type"]
    if message_type == "event" and data["event"]["event"] == "livestream video data":
        image_buffer = data["event"]["buffer"]["data"]
        if not is_h264_complete(image_buffer):
            print(f"Error! incomplete h264: {len(image_buffer)}")
            return
        buffer_bytes = bytes(image_buffer)
        packet = av.Packet(buffer_bytes)
        codec = av.CodecContext.create('h264', 'r')
        frames = codec.decode(packet)
        # Display the image
        for frame in frames:
            image = frame.to_ndarray(format='bgr24')
            # Put the length of the buffer on the image
            cv2.putText(image, f"Buffer Length: {len(image_buffer)}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
            cv2.imshow('Image', image)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
def on_error(ws, error):
    print(f"Error: {error}")
def on_close(ws):
    print("Connection closed")
def on_open(ws):
    print("Connection opened")
    # Send a message to the server
    ws.send(json.dumps({"messageId" : "start_listening", "command": "start_listening"}))  # replace with your command and parameters
    ws.send(json.dumps({"command": "set_api_schema", "schemaVersion" : 20}))
    ws.send(json.dumps({"messageId" : "start_livestream", "command": "device.start_livestream", "serialNumber": "T8410P4223334EBE"}))  # replace with your command and parameters
if __name__ == "__main__":
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp("ws://localhost:3000",  # replace with your server URI
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.on_open = on_open
    ws.run_forever()

问题分析 #

1. CodecContext重复创建：每次解码都新建CodecContext，H264的帧间依赖需要保留解码上下文状态，重复创建会导致无法正确解码后续帧
2. 帧完整性判断错误：要求每个WebSocket包都包含SPS/PPS，实际只有关键帧前会携带SPS/PPS，普通P帧不包含，直接丢弃会导致大量丢帧
3. 未处理分片NALU：WebSocket返回的是分片的H264数据，需要累积缓冲区直到找到完整的NALU边界再解码
4. 回调内阻塞操作：在WebSocket消息回调中调用cv2.imshow和cv2.waitKey，会阻塞消息接收，导致后续帧堆积或丢失

修复后的代码 #

import websocket
import json
import av
import cv2
import threading
from queue import Queue

# 全局变量：解码上下文、帧队列、累积缓冲区
codec = None
frame_queue = Queue(maxsize=30)
h264_buffer = bytearray()
start_code = bytes([0, 0, 0, 1])

def init_codec():
    global codec
    codec = av.CodecContext.create('h264', 'r')
    codec.skip_frame = 'discard'  # 跳过损坏的帧

def extract_nalus(buffer):
    # 从缓冲区中提取完整的NALU
    buffer_bytes = bytes(buffer)
    nalus = []
    start_positions = [i for i in range(len(buffer_bytes)) if buffer_bytes.startswith(start_code, i)]
    
    for i in range(len(start_positions)):
        start = start_positions[i]
        end = start_positions[i+1] if i+1 < len(start_positions) else len(buffer_bytes)
        nalus.append(buffer_bytes[start:end])
    
    # 保留最后一个不完整的NALU到缓冲区
    remaining = buffer_bytes[start_positions[-1]:] if start_positions else buffer_bytes
    return nalus, remaining

def decode_nalus(nalus):
    for nalu in nalus:
        if not nalu:
            continue
        packet = av.Packet(nalu)
        try:
            frames = codec.decode(packet)
            for frame in frames:
                if frame is not None:
                    # 转换为BGR格式并存入队列
                    frame_bgr = frame.to_ndarray(format='bgr24')
                    if not frame_queue.full():
                        frame_queue.put(frame_bgr)
        except Exception as e:
            print(f"解码错误: {e}")

def on_message(ws, message):
    global h264_buffer
    data = json.loads(message)
    message_type = data.get("type")
    
    if message_type == "event" and data["event"]["event"] == "livestream video data":
        image_buffer = data["event"]["buffer"]["data"]
        h264_buffer.extend(image_buffer)
        
        # 提取完整NALU并解码
        nalus, remaining = extract_nalus(h264_buffer)
        h264_buffer = bytearray(remaining)
        decode_nalus(nalus)

def on_error(ws, error):
    print(f"错误: {error}")

def on_close(ws):
    print("连接关闭")

def on_open(ws):
    print("连接已打开")
    init_codec()
    # 发送初始化命令
    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 display_frames():
    # 单独线程处理帧显示，避免阻塞WebSocket回调
    cv2.namedWindow('Eufy Livestream', cv2.WINDOW_NORMAL)
    while True:
        if not frame_queue.empty():
            frame = frame_queue.get()
            cv2.putText(frame, f"Queue Size: {frame_queue.qsize()}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
            cv2.imshow('Eufy Livestream', frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
    cv2.destroyAllWindows()

if __name__ == "__main__":
    websocket.enableTrace(False)
    # 启动帧显示线程
    display_thread = threading.Thread(target=display_frames, daemon=True)
    display_thread.start()
    
    ws = websocket.WebSocketApp(
        "ws://localhost:3000",
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.on_open = on_open
    ws.run_forever()

关键修改说明 #

• 全局CodecContext：初始化一次解码上下文，保留帧间依赖状态，确保后续帧能正确解码
• NALU提取逻辑：累积WebSocket返回的分片数据，提取完整的NALU单元再解码，避免分片导致的解码失败
• 帧队列+单独显示线程：将解码后的帧存入队列，由单独线程处理显示，避免阻塞WebSocket消息接收
• 错误处理：添加解码异常捕获，跳过损坏帧，保证播放流畅性

内容的提问来源于stack exchange，提问作者yarin Cohen