如何为 tensorflow.keras.preprocessing.text_dataset_from_directory 使用多个输入

Question

所以我正在训练一个 CNN，它接收输入的两个图像并返回一个值作为 GPU 上的输出。由于我有很多图像，因此为了将数据分批提供，我正在使用 tf.keras.preprocessing.text_dataset_from_directory 创建一个针对 GPU 进行了优化的 tf.Dataset 对象。

所以基本上我的输入目录是

Class_1/
  Class1_1/
     image1.png
     image2.png
  Class1_2/
     image3.png
     image4.png
...
Class_2/
  Class2_1/
     image1.png
     image2.png
  Class2_2/
     image3.png
     image4.png

默认函数只适用于以下结构

Class_1/
      image1.png
      image2.png
      image3.png
      image4.png
    ...
Class_2/
      image1.png
      image2.png
      image3.png
      image4.png

任何帮助将不胜感激。

Answer 1

我认为您的意思是 image_dataset_from_directory，因为您正在加载图像而不是文本数据。无论哪种方式，您都无法从这些辅助函数生成具有多个输入的批次，您可以看到 from the documentation 已定义返回形状：

一个 tf.data.Dataset 对象。

• 如果 label_mode 为 None，它会生成 float32 形状张量（batch_size、image_size[0]、image_size[1]、num_channels）、编码图像（有关 num_channels 的规则，请参见下文）。
• 否则，它会生成一个元组（图像、标签），其中图像具有形状（batch_size、image_size[0]、image_size[1]、num_channels），标签遵循下述格式。

您将需要编写自己的自定义生成器函数，该函数生成从数据目录加载的多个输入，然后使用自定义生成器调用 fit 并将 kwarg validation_data 传递给生成验证数据的单独生成器。 （注意：在某些旧版本的 Keras 中，您可能需要 fit_generator 而不是 fit）。

下面是一些辅助函数模块的示例，它可以从某些目录中读取图像并将它们呈现为训练中的多图像输入。

def _generate_batch(training):
    in1s, in2s, labels = [], [], []
    batch_tuples = _sample_batch_of_paths(training)
    for input1_path, input2_path in batch_tuples:
        # skip any exception so that image GPU batch loading isn't
        # disrupted and any faulty image is just skipped.
        try:
            in1_tmp = _load_image(
                os.path.join(INPUT1_PATH_PREFIX, input1_path),
            )
            in2_tmp = _load_image(
                os.path.join(INPUT2_PATH_PREFIX, input2_path),
            )
        except Exception as exc:
            print("Unhandled exception during image batch load. Skipping...")
            print(str(exc))
            continue
        # if no exception, both images loaded so both are added to batch.
        in1s.append(in1_tmp)
        in2s.append(in2_tmp)
        # Whatever your custom logic is to determine the label for the pair.
        labels.append(
            _label_calculation_helper(input1_path, input2_path)
        )
    in1s, in2s = map(skimage.io.concatenate_images, [in1s, in2s])
    # could also add a singleton channel dimension for grayscale images.
    # in1s = in1s[:, :, :, None]
    return [in1s, in2s], labels


def _make_generator(training=True):
    while True:
        yield _generate_batch(training)


def make_generators():
    return _make_generator(training=True), _make_generator(training=False)

helper _load_image 可能是这样的：

def _load_image(path, is_gray=False):
    tmp = skimage.io.imread(path)
    if is_gray:
        tmp = skimage.util.img_as_float(skimage.color.rgb2gray(tmp))
    else:
        tmp = skimage.util.img_as_float(skimage.color.gray2rgb(tmp))
        if tmp.shape[-1] == 4:
            tmp = skimage.color.rgba2rgb(tmp)
    # Do other stuff here - resizing, clipping, etc.
    return tmp

从磁盘上列出的一组路径中对批次进行采样的辅助函数可能是这样的：

@lru_cache(1)
def _load_and_split_input_paths():
    training_in1s, testing_in1s = train_test_split(
        os.listdir(INPUT1_PATH_PREFIX),
        test_size=TEST_SIZE,
        random_state=RANDOM_SEED
    )
    training_in2s, testing_in2s = train_test_split(
        os.listdir(INPUT2_PATH_PREFIX),
        test_size=TEST_SIZE,
        random_state=RANDOM_SEED
    )
    return training_in1s, testing_in1s, training_in2s, testing_in2s


def _sample_batch_of_paths(training):
    training_in1s, testing_in1s, training_in2s, testing_in2s = _load_and_split_input_paths()
    if training:
        return list(zip(
            random.sample(training_in1s, BATCH_SIZE),
            random.sample(training_in2s, BATCH_SIZE)
        ))
    else:
        return list(zip(
            random.sample(testing_in1s, BATCH_SIZE),
            random.sample(testing_in2s, BATCH_SIZE)
        ))

这将从“输入 1”目录中随机抽取图像，并将它们与“输入 2”目录中的随机样本配对。显然，在您的用例中，您需要对此进行更改，以便根据定义其配对和标签的文件结构确定性地提取数据。

最后一旦你想用这个，你可以调用训练代码如：

training_generator, testing_generator = make_generators()
try:
    some_compiled_model.fit(
        training_generator,
        epochs=EPOCHS,
        validation_data=testing_generator,
        callbacks=[...],
        verbose=VERBOSE,
        steps_per_epoch=STEPS_PER_EPOCH,
        validation_steps=VALIDATION_STEPS,
    )
except KeyboardInterrupt:
    pass