Python 处理来自 Eufy 服务器的直播流的 H264 帧

Question

我目前正在使用 Eufy Security WebSocket Server，这是一个围绕 eufy-security-client 库构建的服务器包装器，支持通过 WebSocket 接口进行访问。我开发了一个 python 版本的客户端，我尝试使用

device.start_livestream

命令显示实时流，如 此处概述。

简而言之，Web 服务器不断返回 H264 格式的视频帧缓冲区。同时，我在 Python 脚本中阅读了此内容并尝试将其呈现在 GUI 上。但是，我遇到了丢失大量帧的问题。这可能是由于连续帧的压缩所致，这一过程称为帧间压缩。

解决方案可能在于实现缓冲或数据包重组逻辑，以确保处理完整的帧。尽管尝试了很多方法，探索了很多方法，我还是想不通。

这是我的Python完整代码：

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()

Answer 1

您的方法 is_h264_complete 在这些上下文中对我来说看起来是错误的，这个函数看起来像是检查 I 帧、PPS 和 SPS 等流中的同步点以初始化解码器。

目前您仅解码此同步点（I 帧），不解码 P、B 帧。您可以删除此函数并在 on_message 回调之外创建 Codex 上下文。使用连接回调左右，因为我假设您在消息回调中获得多个 h264 有效负载。 目前，您在每条消息上创建一个新的上下文，例如解码器，这太疯狂了。

创建编解码器上下文一次，并在消息中通过 av.Packet 将数据馈送到解码器并调用解码。 FFmpeg 足够聪明，可以解析字节流。

在断开连接时，您可以关闭，例如释放编解码器上下文。