如何序列化PyAV.packet跨机传输并还原？附元数据打包方案咨询

一、还原PyAV Packet对象 #

你之前用update()方法无效是因为PyAV的Packet.update()是用于从另一个Packet实例复制数据，而非从原始字节填充内容。正确的做法是同时序列化Packet的原始字节和关键元数据，接收端再重新构造Packet并恢复元数据——仅靠字节无法还原Packet的时序（pts/dts）、流索引等关键信息，这些是解码和播放必需的。

发送端（序列化Packet） #

import av
import msgpack

# 假设packet是从RTSP获取的PyAV Packet
packet_info = {
    "data": bytes(packet),
    "pts": packet.pts,
    "dts": packet.dts,
    "duration": packet.duration,
    "stream_index": packet.stream_index,
    "flags": packet.flags,
    "pos": packet.pos
}

# 序列化为可传输的bytes
serialized_data = msgpack.packb(packet_info)
# 后续通过socket/MQ发送serialized_data即可

接收端（还原Packet） #

import av
import msgpack

# 接收从socket/MQ获取的serialized_data
packet_info = msgpack.unpackb(serialized_data)

# 从字节构造Packet
restored_packet = av.packet.Packet(packet_info["data"])
# 恢复关键元数据
restored_packet.pts = packet_info["pts"]
restored_packet.dts = packet_info["dts"]
restored_packet.duration = packet_info["duration"]
restored_packet.stream_index = packet_info["stream_index"]
restored_packet.flags = packet_info["flags"]
restored_packet.pos = packet_info["pos"]

# 若需解码，需将该Packet关联到对应流的codec上下文（需与发送端一致）

二、Packet+逐帧元数据传输方案 #

除了上述“元数据+Packet字节”分离序列化的方式，还有以下两种常用方案：

方案1：自定义轻量字节协议 #

自己定义传输格式，按固定顺序封装元数据和Packet字节，无需依赖第三方序列化库：

发送端

import json
import struct

# 假设预处理元数据
metadata = {"frame_id": "frame_001", "detect_results": ["person", "car"]}
metadata_bytes = json.dumps(metadata).encode("utf-8")
packet_bytes = bytes(packet)

# 构造传输包：[元数据长度(4字节大端)] + [元数据字节] + [Packet长度(4字节大端)] + [Packet字节]
transmit_data = struct.pack(">I", len(metadata_bytes)) + metadata_bytes
transmit_data += struct.pack(">I", len(packet_bytes)) + packet_bytes

接收端

import json
import struct

# 按顺序读取数据（需确保socket接收完整数据，可做循环读取）
# 读取元数据长度
metadata_len = struct.unpack(">I", recv(4))[0]
# 读取元数据
metadata_bytes = recv(metadata_len)
metadata = json.loads(metadata_bytes.decode("utf-8"))
# 读取Packet长度
packet_len = struct.unpack(">I", recv(4))[0]
# 读取Packet字节并还原
packet_bytes = recv(packet_len)
restored_packet = av.packet.Packet(packet_bytes)
# 恢复Packet元数据（同前文方法）

方案2：H264 SEI消息嵌入元数据 #

将元数据嵌入H264的SEI（Supplemental Enhancement Information）消息中，作为视频流的一部分传输，适合需将元数据与视频帧强绑定的场景：

发送端（构造SEI Packet）

import av
import json

def create_h264_sei_packet(metadata, pts, dts, stream_index):
    # 将元数据转为字节
    metadata_bytes = json.dumps(metadata).encode("utf-8")
    # 构造SEI payload：payload type(6=自定义用户数据) + payload长度 + 元数据
    # 注：payload长度需用UE-V编码，此处简化为单字节（仅支持长度<255的元数据）
    sei_payload = bytes([6]) + bytes([len(metadata_bytes)]) + metadata_bytes
    # 构造H264 NALU：起始码 + NALU头(type=6) + SEI payload
    nalu = b"\x00\x00\x00\x01" + bytes([0x60 | 6]) + sei_payload
    # 创建SEI Packet并设置时序信息
    sei_packet = av.packet.Packet(nalu)
    sei_packet.pts = pts
    sei_packet.dts = dts
    sei_packet.stream_index = stream_index
    return sei_packet

# 构造SEI Packet并与视频Packet一起发送
metadata = {"frame_id": "frame_001", "preprocess_info": "blur_applied"}
sei_packet = create_h264_sei_packet(metadata, packet.pts, packet.dts, packet.stream_index)

# 先发送SEI Packet的bytes，再发送视频Packet的bytes

接收端（解析SEI）

import av
import json

# 接收字节并还原Packet
received_packet = av.packet.Packet(received_bytes)
packet_data = bytes(received_packet)

# 判断是否为SEI NALU
if len(packet_data) >= 5 and packet_data[:4] == b"\x00\x00\x00\x01":
    nalu_type = packet_data[4] & 0x1F
    if nalu_type == 6:
        # 解析SEI元数据
        sei_payload = packet_data[5:]
        payload_type = sei_payload[0]
        payload_size = sei_payload[1]
        metadata_bytes = sei_payload[2:2+payload_size]
        metadata = json.loads(metadata_bytes.decode("utf-8"))
        # 处理元数据
    else:
        # 处理视频帧Packet
        restored_packet = received_packet
        # 恢复元数据（同前文方法）

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