Python图片存储和访问的三种方式是什么

import numpy as npimport picklefrom pathlib import Path# 文件路径data_dir = Path("data/cifar-10-batches-py/")# 解码功能def unpickle(file):    with open(file, "rb") as fo:        dict = pickle.load(fo, encoding="bytes")    return dictimages, labels = [], []for batch in data_dir.glob("data_batch_*"):    batch_data = unpickle(batch)    for i, flat_im in enumerate(batch_data[b"data"]):        im_channels = []        # 每个图像都是扁平化的，通道按 R, G, B 的顺序排列          for j in range(3):            im_channels.append(                flat_im[j * 1024 : (j + 1) * 1024].reshape((32, 32))            )        # 重建原始图像        images.append(np.dstack((im_channels)))        # 保存标签        labels.append(batch_data[b"labels"][i])print("加载 CIFAR-10 训练集:")print(f" - np.shape(images)     {np.shape(images)}")print(f" - np.shape(labels)     {np.shape(labels)}")

pip install Pillow

pip install lmdb

pip install h6py

from pathlib import Pathdisk_dir = Path("data/disk/")lmdb_dir = Path("data/lmdb/")hdf5_dir = Path("data/hdf5/")

disk_dir.mkdir(parents=True, exist_ok=True)lmdb_dir.mkdir(parents=True, exist_ok=True)hdf5_dir.mkdir(parents=True, exist_ok=True)

from PIL import Imageimport csvdef store_single_disk(image, image_id, label):    """ 将单个图像作为 .png 文件存储在磁盘上。        参数:        ---------------        image       图像数组， (32, 32, 3) 格式        image_id    图像的整数唯一 ID        label       图像标签    """    Image.fromarray(image).save(disk_dir / f"{image_id}.png")    with open(disk_dir / f"{image_id}.csv", "wt") as csvfile:        writer = csv.writer(            csvfile, delimiter=" ", quotechar="|", quoting=csv.QUOTE_MINIMAL        )        writer.writerow([label])

class CIFAR_Image:    def __init__(self, image, label):        self.channels = image.shape[2]        self.size = image.shape[:2]        self.image = image.tobytes()        self.label = label    def get_image(self):        """ 将图像作为 numpy 数组返回 """        image = np.frombuffer(self.image, dtype=np.uint8)        return image.reshape(*self.size, self.channels)

import lmdbimport pickledef store_single_lmdb(image, image_id, label):    """ 将单个图像存储到 LMDB        参数:        ---------------        image       图像数组， (32, 32, 3) 格式        image_id    图像的整数唯一 ID        label       图像标签    """    map_size = image.nbytes * 10    # Create a new LMDB environment    env = lmdb.open(str(lmdb_dir / f"single_lmdb"), map_size=map_size)    # Start a new write transaction    with env.begin(write=True) as txn:        # All key-value pairs need to be strings        value = CIFAR_Image(image, label)        key = f"{image_id:08}"        txn.put(key.encode("ascii"), pickle.dumps(value))    env.close()

import h6pydef store_single_hdf5(image, image_id, label):    """ 将单个图像存储到 HDF5 文件        参数:        ---------------        image       图像数组， (32, 32, 3) 格式        image_id    图像的整数唯一 ID        label       图像标签    """    # 创建一个新的 HDF5 文件    file = h6py.File(hdf5_dir / f"{image_id}.h6", "w")    # 在文件中创建数据集    dataset = file.create_dataset(        "image", np.shape(image), h6py.h6t.STD_U8BE, data=image    )    meta_set = file.create_dataset(        "meta", np.shape(label), h6py.h6t.STD_U8BE, data=label    )    file.close()

_store_single_funcs = dict(    disk=store_single_disk,     lmdb=store_single_lmdb,     hdf5=store_single_hdf5)

from timeit import timeitstore_single_timings = dict()for method in ("disk", "lmdb", "hdf5"):    t = timeit(        "_store_single_funcs[method](image, 0, label)",        setup="image=images[0]; label=labels[0]",        number=1,        globals=globals(),    )    store_single_timings[method] = t    print(f"存储方法: {method}, 使用耗时: {t}")

 store_many_disk(images, labels):    """ 参数:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    num_images = len(images)    # 一张一张保存所有图片    for i, image in enumerate(images):        Image.fromarray(image).save(disk_dir / f"{i}.png")    # 将所有标签保存到 csv 文件    with open(disk_dir / f"{num_images}.csv", "w") as csvfile:        writer = csv.writer(            csvfile, delimiter=" ", quotechar="|", quoting=csv.QUOTE_MINIMAL        )        for label in labels:            writer.writerow([label])

def store_many_lmdb(images, labels):    """ 参数:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    num_images = len(images)    map_size = num_images * images[0].nbytes * 10    # 为所有图像创建一个新的 LMDB 数据库    env = lmdb.open(str(lmdb_dir / f"{num_images}_lmdb"), map_size=map_size)    # 在一个事务中写入所有图像    with env.begin(write=True) as txn:        for i in range(num_images):            # 所有键值对都必须是字符串            value = CIFAR_Image(images[i], labels[i])            key = f"{i:08}"            txn.put(key.encode("ascii"), pickle.dumps(value))    env.close()

def store_many_hdf5(images, labels):    """ 参数:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    num_images = len(images)    # 创建一个新的 HDF5 文件    file = h6py.File(hdf5_dir / f"{num_images}_many.h6", "w")    # 在文件中创建数据集    dataset = file.create_dataset(        "images", np.shape(images), h6py.h6t.STD_U8BE, data=images    )    meta_set = file.create_dataset(        "meta", np.shape(labels), h6py.h6t.STD_U8BE, data=labels    )    file.close()

cutoffs = [10, 100, 1000, 10000, 100000]images = np.concatenate((images, images), axis=0)labels = np.concatenate((labels, labels), axis=0)# 确保有 100,000 个图像和标签print(np.shape(images))print(np.shape(labels))

_store_many_funcs = dict(    disk=store_many_disk, lmdb=store_many_lmdb, hdf5=store_many_hdf5)from timeit import timeitstore_many_timings = {"disk": [], "lmdb": [], "hdf5": []}for cutoff in cutoffs:    for method in ("disk", "lmdb", "hdf5"):        t = timeit(            "_store_many_funcs[method](images_, labels_)",            setup="images_=images[:cutoff]; labels_=labels[:cutoff]",            number=1,            globals=globals(),        )        store_many_timings[method].append(t)        # 打印出方法、截止时间和使用时间        print(f"Method: {method}, Time usage: {t}")

import matplotlib.pyplot as pltdef plot_with_legend(    x_range, y_data, legend_labels, x_label, y_label, title, log=False):    """ 参数:        --------------        x_range         包含 x 数据的列表        y_data          包含 y 值的列表        legend_labels   字符串图例标签列表        x_label         x 轴标签        y_label         y 轴标签    """    plt.style.use("seaborn-whitegrid")    plt.figure(figsize=(10, 7))    if len(y_data) != len(legend_labels):        raise TypeError(            "数据集的数量与标签的数量不匹配"        )    all_plots = []    for data, label in zip(y_data, legend_labels):        if log:            temp, = plt.loglog(x_range, data, label=label)        else:            temp, = plt.plot(x_range, data, label=label)        all_plots.append(temp)    plt.title(title)    plt.xlabel(x_label)    plt.ylabel(y_label)    plt.legend(handles=all_plots)    plt.show()# Getting the store timings data to displaydisk_x = store_many_timings["disk"]lmdb_x = store_many_timings["lmdb"]hdf5_x = store_many_timings["hdf5"]plot_with_legend(    cutoffs,    [disk_x, lmdb_x, hdf5_x],    ["PNG files", "LMDB", "HDF5"],    "Number of images",    "Seconds to store",    "Storage time",    log=False,)plot_with_legend(    cutoffs,    [disk_x, lmdb_x, hdf5_x],    ["PNG files", "LMDB", "HDF5"],    "Number of images",    "Seconds to store",    "Log storage time",    log=True,)

def read_single_disk(image_id):    """ 参数:        ---------------        image_id    图像的整数唯一 ID        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    image = np.array(Image.open(disk_dir / f"{image_id}.png"))    with open(disk_dir / f"{image_id}.csv", "r") as csvfile:        reader = csv.reader(            csvfile, delimiter=" ", quotechar="|", quoting=csv.QUOTE_MINIMAL        )        label = int(next(reader)[0])    return image, label

def read_single_lmdb(image_id):    """ 参数:        ---------------        image_id    图像的整数唯一 ID        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    # 打开 LMDB 环境    env = lmdb.open(str(lmdb_dir / f"single_lmdb"), readonly=True)    # 开始一个新的事务    with env.begin() as txn:        # 进行编码        data = txn.get(f"{image_id:08}".encode("ascii"))        # 加载的 CIFAR_Image 对象        cifar_image = pickle.loads(data)        # 检索相关位        image = cifar_image.get_image()        label = cifar_image.label    env.close()    return image, label

def read_single_hdf5(image_id):    """ 参数:        ---------------        image_id    图像的整数唯一 ID        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    # 打开 HDF5 文件    file = h6py.File(hdf5_dir / f"{image_id}.h6", "r+")    image = np.array(file["/image"]).astype("uint8")    label = int(np.array(file["/meta"]).astype("uint8"))    return image, label

from timeit import timeitread_single_timings = dict()for method in ("disk", "lmdb", "hdf5"):    t = timeit(        "_read_single_funcs[method](0)",        setup="image=images[0]; label=labels[0]",        number=1,        globals=globals(),    )    read_single_timings[method] = t	print(f"读取方法: {method}, 使用耗时: {t}")

def read_many_disk(num_images):    """ 参数:        ---------------        num_images   要读取的图像数量        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    images, labels = [], []    # 循环遍历所有ID，一张一张地读取每张图片    for image_id in range(num_images):        images.append(np.array(Image.open(disk_dir / f"{image_id}.png")))    with open(disk_dir / f"{num_images}.csv", "r") as csvfile:        reader = csv.reader(            csvfile, delimiter=" ", quotechar="|", quoting=csv.QUOTE_MINIMAL        )        for row in reader:            labels.append(int(row[0]))    return images, labels

def read_many_lmdb(num_images):    """ 参数:        ---------------        num_images   要读取的图像数量        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    images, labels = [], []    env = lmdb.open(str(lmdb_dir / f"{num_images}_lmdb"), readonly=True)    # 开始一个新的事务    with env.begin() as txn:        # 在一个事务中读取，也可以拆分成多个事务分别读取        for image_id in range(num_images):            data = txn.get(f"{image_id:08}".encode("ascii"))            # CIFAR_Image 对象，作为值存储            cifar_image = pickle.loads(data)            # 检索相关位            images.append(cifar_image.get_image())            labels.append(cifar_image.label)    env.close()    return images, labels

def read_many_hdf5(num_images):    """ 参数:        ---------------        num_images   要读取的图像数量        返回结果:        ---------------        images       图像数组 (N, 32, 32, 3) 格式        labels       标签数组 (N，1) 格式    """    images, labels = [], []    # 打开 HDF5 文件    file = h6py.File(hdf5_dir / f"{num_images}_many.h6", "r+")    images = np.array(file["/images"]).astype("uint8")    labels = np.array(file["/meta"]).astype("uint8")    return images, labels_read_many_funcs = dict(    disk=read_many_disk, lmdb=read_many_lmdb, hdf5=read_many_hdf5)

from timeit import timeitread_many_timings = {"disk": [], "lmdb": [], "hdf5": []}for cutoff in cutoffs:    for method in ("disk", "lmdb", "hdf5"):        t = timeit(            "_read_many_funcs[method](num_images)",            setup="num_images=cutoff",            number=1,            globals=globals(),        )        read_many_timings[method].append(t)        # Print out the method, cutoff, and elapsed time        print(f"读取方法: {method}, No. images: {cutoff}, 耗时: {t}")

plot_with_legend(    cutoffs,    [disk_x_r, lmdb_x_r, hdf5_x_r, disk_x, lmdb_x, hdf5_x],    [        "Read PNG",        "Read LMDB",        "Read HDF5",        "Write PNG",        "Write LMDB",        "Write HDF5",    ],    "Number of images",    "Seconds",    "Log Store and Read Times",    log=False,)