python - 在 Tensorboard 上显示图像(通过 Keras)

Question

我的 X_test 是 128x128x3 图像，我的 Y_test 是 512x512x3 图像。我想在每个纪元之后展示输入 (X_test) 的外观、预期输出 (Y_test) 的外观，以及实际输出的外观。到目前为止，我只知道如何在 Tensorboard 中添加前 2 个。下面是调用回调的代码:

model.fit(X_train,
          Y_train,
          epochs=epochs,
          verbose=2,
          shuffle=False,
          validation_data=(X_test, Y_test),
          batch_size=batch_size,
          callbacks=get_callbacks())

这是回调的代码:

import tensorflow as tf
from keras.callbacks import Callback
from keras.callbacks import TensorBoard

import io
from PIL import Image

from constants import batch_size


def get_callbacks():
    tbCallBack = TensorBoard(log_dir='./logs',
                             histogram_freq=1,
                             write_graph=True,
                             write_images=True,
                             write_grads=True,
                             batch_size=batch_size)

    tbi_callback = TensorBoardImage('Image test')

    return [tbCallBack, tbi_callback]


def make_image(tensor):
    """
    Convert an numpy representation image to Image protobuf.
    Copied from https://github.com/lanpa/tensorboard-pytorch/
    """
    height, width, channel = tensor.shape
    print(tensor)
    image = Image.fromarray(tensor.astype('uint8'))  # TODO: maybe float ?

    output = io.BytesIO()
    image.save(output, format='JPEG')
    image_string = output.getvalue()
    output.close()

    return tf.Summary.Image(height=height,
                            width=width,
                            colorspace=channel,
                            encoded_image_string=image_string)


class TensorBoardImage(Callback):
    def __init__(self, tag):
        super().__init__()
        self.tag = tag

    def on_epoch_end(self, epoch, logs={}):
        # Load image
        img_input = self.validation_data[0][0]  # X_train
        img_valid = self.validation_data[1][0]  # Y_train

        print(self.validation_data[0].shape)  # (8, 128, 128, 3)
        print(self.validation_data[1].shape)  # (8, 512, 512, 3)

        image = make_image(img_input)
        summary = tf.Summary(value=[tf.Summary.Value(tag=self.tag, image=image)])
        writer = tf.summary.FileWriter('./logs')
        writer.add_summary(summary, epoch)
        writer.close()

        image = make_image(img_valid)
        summary = tf.Summary(value=[tf.Summary.Value(tag=self.tag, image=image)])
        writer = tf.summary.FileWriter('./logs')
        writer.add_summary(summary, epoch)
        writer.close()

        return

我想知道在哪里/如何获得网络的实际输出。

我遇到的另一个问题是，这是正在移植到 TensorBoard 中的图像之一的示例:

[[[0.10909907 0.09341043 0.08224604]
  [0.11599099 0.09922747 0.09138277]
  [0.15596421 0.13087936 0.11472746]
  ...
  [0.87589591 0.72773653 0.69428956]
  [0.87006552 0.7218123  0.68836991]
  [0.87054225 0.72794635 0.6967475 ]]

 ...

 [[0.26142332 0.16216267 0.10314116]
  [0.31526875 0.18743924 0.12351286]
  [0.5499796  0.35461449 0.24772873]
  ...
  [0.80937942 0.62956016 0.53784871]
  [0.80906054 0.62843601 0.5368183 ]
  [0.81046278 0.62453899 0.53849678]]]

这就是为什么我的 image = Image.fromarray(tensor.astype('uint8')) 行可能生成的图像看起来与实际输出完全不同的原因吗？这是来自 TensorBoard 的示例:

我确实尝试了 .astype('float64') 但它引发了错误，因为它显然不是受支持的类型。

无论如何，我不确定这是否真的是问题所在，因为我在 TensorBoard 中显示的其余图像都只是白色/灰色/黑色方 block (就在那里，conv2D_7，实际上是我网络的最后一层，因此应该显示输出的实际图像，不是吗？):

最终，我想要这样的东西，我已经在通过 matplot 训练后展示了它:

最后，我想说明这个回调需要很长时间才能处理。有没有更有效的方法来做到这一点？它几乎使我的训练时间增加了一倍(可能是因为它需要先将 numpy 转换为图像，然后再将它们保存到 TensorBoard 日志文件中)。

Answer 1

以下代码接受模型输入、模型输出和地面实况，并保存到 Tensorboard。该模型是分割的，因此每个样本 3 个图像。

代码非常简单明了。但还是有几点解释:-

make_image_tensor - 该方法转换 numpy 图像并创建张量以保存在 tensorboard 摘要中。

TensorBoardWriter - 不是必需的，但最好将 Tensorboard 功能与其他模块分开。允许重复使用。

ModelDiagonoser - 采用生成器并预测 self.model(由 Keras 设置为所有回调)的类。 ModelDiagonoser 获取输入、输出和真实数据并传递给 Tensorboard 以保存图像。

import os

import io
import numpy as np
import tensorflow as tf
from PIL import Image
from tensorflow.keras.callbacks import Callback

# Depending on your keras version use one of the following:
# from tensorflow.keras.engine.training import GeneratorEnqueuer, Sequence, OrderedEnqueuer
from tensorflow.keras.utils import GeneratorEnqueuer, Sequence, OrderedEnqueuer


def make_image_tensor(tensor):
    """
    Convert an numpy representation image to Image protobuf.
    Adapted from https://github.com/lanpa/tensorboard-pytorch/
    """
    if len(tensor.shape) == 3:
        height, width, channel = tensor.shape
    else:
        height, width = tensor.shape
        channel = 1
    tensor = tensor.astype(np.uint8)
    image = Image.fromarray(tensor)
    output = io.BytesIO()
    image.save(output, format='PNG')
    image_string = output.getvalue()
    output.close()
    return tf.Summary.Image(height=height,
                            width=width,
                            colorspace=channel,
                            encoded_image_string=image_string)


class TensorBoardWriter:

    def __init__(self, outdir):
        assert (os.path.isdir(outdir))
        self.outdir = outdir
        self.writer = tf.summary.FileWriter(self.outdir,
                                            flush_secs=10)

    def save_image(self, tag, image, global_step=None):
        image_tensor = make_image_tensor(image)
        self.writer.add_summary(tf.Summary(value=[tf.Summary.Value(tag=tag, image=image_tensor)]),
                                global_step)

    def close(self):
        """
        To be called in the end
        """
        self.writer.close()


class ModelDiagonoser(Callback):

    def __init__(self,
                 data_generator,
                 batch_size,
                 num_samples,
                 output_dir,
                 normalization_mean):
        self.batch_size = batch_size
        self.num_samples = num_samples
        self.tensorboard_writer = TensorBoardWriter(output_dir)
        self.normalization_mean = normalization_mean
        is_sequence = isinstance(data_generator, Sequence)
        if is_sequence:
            self.enqueuer = OrderedEnqueuer(data_generator,
                                            use_multiprocessing=True,
                                            shuffle=False)
        else:
            self.enqueuer = GeneratorEnqueuer(data_generator,
                                              use_multiprocessing=True)
        self.enqueuer.start(workers=4, max_queue_size=4)

    def on_epoch_end(self, epoch, logs=None):
        output_generator = self.enqueuer.get()
        steps_done = 0
        total_steps = int(np.ceil(np.divide(self.num_samples, self.batch_size)))
        sample_index = 0
        while steps_done < total_steps:
            generator_output = next(output_generator)
            x, y = generator_output[:2]
            y_pred = self.model.predict(x)
            y_pred = np.argmax(y_pred, axis=-1)
            y_true = np.argmax(y, axis=-1)

            for i in range(0, len(y_pred)):
                n = steps_done * self.batch_size + i
                if n >= self.num_samples:
                    return
                img = np.squeeze(x[i, :, :, :])
                img = 255. * (img + self.normalization_mean)  # mean is the training images normalization mean
                img = img[:, :, [2, 1, 0]]  # reordering of channels

                pred = y_pred[i]
                pred = pred.reshape(img.shape[0:2])

                ground_truth = y_true[i]
                ground_truth = ground_truth.reshape(img.shape[0:2])

                self.tensorboard_writer.save_image("Epoch-{}/{}/x"
                                                   .format(epoch, sample_index), img)
                self.tensorboard_writer.save_image("Epoch-{}/{}/y"
                                                   .format(epoch, sample_index), ground_truth)
                self.tensorboard_writer.save_image("Epoch-{}/{}/y_pred"
                                                   .format(epoch, sample_index), pred)
                sample_index += 1

            steps_done += 1

    def on_train_end(self, logs=None):
        self.enqueuer.stop()
        self.tensorboard_writer.close()