如何将4张PNG图片序列化到Flatbuffer数组并正确反序列化？

问题背景 #

需要将images文件夹中的4张PNG序列化到Flatbuffer，存入数组后发送到远端，但现有代码反序列化后仅得到异常数值，无法还原原始图片。

错误原因分析 #

原Schema中image表的pixelData定义为单个[ubyte]（字节数组），但序列化脚本错误地将多个图片字节数组的Flatbuffer向量偏移量存入该数组，导致反序列化时读取到的是偏移数值而非原始图片字节数据，自然无法还原PNG。

正确解决方案 #

1. 修改Flatbuffer Schema #

重新定义Schema，用Image表存储单张图片的元数据和原始字节，再用ImageBatch作为根类型存储多张图片的数组：

namespace imageBuffer;

table Image {
    width: int;            // 图片宽度
    height: int;           // 图片高度
    pixelData: [ubyte];    // 单张图片的原始二进制数据
}

table ImageBatch {
    images: [Image];       // 多张图片的数组
}

root_type ImageBatch;

执行命令生成Python代码：

flatc --python image_schema.fbs

2. 序列化脚本 #

读取每张图片的原始字节和实际宽高，构建Image对象后存入ImageBatch数组：

import flatbuffers
import imageBuffer.Image as Image
import imageBuffer.ImageBatch as ImageBatch
import os
from PIL import Image as PILImage  # 命名避免冲突

def main():
    folder = "images"
    image_offsets = []

    # 初始化Builder，增大初始容量适配图片数据
    builder = flatbuffers.Builder(1024 * 1024)

    for pic in os.listdir(folder):
        if pic.lower().endswith((".png", ".jpg")):
            path = os.path.join(folder, pic)
            # 读取图片原始二进制数据
            with open(path, 'rb') as f:
                pixel_data = f.read()
            # 获取图片实际宽高
            with PILImage.open(path) as img:
                width, height = img.size

            # 构建pixelData字节向量
            pixel_vec = builder.CreateByteVector(pixel_data)
            # 构建单个Image对象
            Image.Start(builder)
            Image.AddWidth(builder, width)
            Image.AddHeight(builder, height)
            Image.AddPixelData(builder, pixel_vec)
            image_offset = Image.End(builder)
            image_offsets.append(image_offset)

    # 构建images数组（Flatbuffer数组需反向插入）
    ImageBatch.StartImagesVector(builder, len(image_offsets))
    for offset in reversed(image_offsets):
        builder.PrependUOffsetTRelative(offset)
    images_vec = builder.EndVector(len(image_offsets))

    # 构建根对象ImageBatch
    ImageBatch.Start(builder)
    ImageBatch.AddImages(builder, images_vec)
    batch_offset = ImageBatch.End(builder)

    builder.Finish(batch_offset)
    serialized_data = builder.Output()

    # 写入二进制文件
    bin_file = "/home/root/serialized_images.bin"
    with open(bin_file, 'wb') as f:
        f.write(serialized_data)

if __name__ == "__main__":
    main()

3. 反序列化脚本 #

解析Flatbuffer数据，提取每张图片的字节数据并还原为原始PNG：

import imageBuffer.ImageBatch as ImageBatch
import imageBuffer.Image as Image
import os

def main():
    bin_file = "/home/root/serialized_images.bin"
    with open(bin_file, 'rb') as f:
        serialized_data = f.read()

    # 解析根对象ImageBatch
    batch = ImageBatch.ImageBatch.GetRootAsImageBatch(serialized_data, 0)
    image_count = batch.ImagesLength()

    # 创建还原图片的输出目录
    output_dir = "restored_images"
    os.makedirs(output_dir, exist_ok=True)

    for i in range(image_count):
        img = batch.Images(i)
        width = img.Width()
        height = img.Height()
        # 提取完整的图片字节数据
        pixel_data = bytearray()
        for j in range(img.PixelDataLength()):
            pixel_data.append(img.PixelData(j))
        
        # 保存为原始图片
        output_path = os.path.join(output_dir, f"restored_image_{i+1}.png")
        with open(output_path, 'wb') as f:
            f.write(pixel_data)
        print(f"已还原图片: {output_path}, 宽: {width}, 高: {height}")

if __name__ == "__main__":
    main()

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