如何使用TensorFlow阅读器和队列同时读取两个文件？

Question

我的训练集包含两种文件:训练图像,文件名如"1.png",标签文件,名称如"1.label.txt".

我在教程中发现了一些Queue和Reader的用法,如下所示:

filename_queue = tf.train.string_input_producer(filenames)
result.key, value = reader.read(filename_queue)

但是,因为我的训练集包含两种文件,一种对应一种.我怎样才能像上面的代码一样使用Queue和Reader？

---

编辑

我正在考虑使用一个包含基本名称的队列来提供给另外两个队列,分别是图像和标签.像这样的代码:

with tf.Session() as sess:
  base_name_queue = tf.train.string_input_producer(['image_names'], num_epochs=20)
  base_name = base_name_queue.dequeue()
  image_name = base_name + ".png"
  image_name_queue = data_flow_ops.FIFOQueue(32, image_name.dtype.base_dtype)
  image_name_queue.enqueue([image_name])
  x = image_name_queue.dequeue()
  print_op = tf.Print(image_name, [image_name])

  qr = tf.train.QueueRunner(base_name_queue, [base_name_queue] * 4)
  coord = tf.train.Coordinator()
  enqueue_threads = qr.create_threads(sess, coord=coord, start=True)

  for step in range(1000000):
    if coord.should_stop():
      break
    print(sess.run(print_op))

  coord.request_stop()
  coord.join(enqueue_threads)

但是运行此代码会导致错误:

TypeError:Fetch参数有无效类型,必须是字符串或Tensor.(无法将FIFOQueue转换为Tensor或Operation.)

并且错误指向此行:

coord.join(enqueue_threads)

我想我必须误解TensorFlow队列的工作原理.

Answer 1

我找到了解决问题的方法.我想在这里发帖回答而不是删除我的问题,希望这将有助于TensorFlow新手.

答案包含两部分:

第1部分:如何使用TensorFlow的队列逐对读取文件

解决方案很简单:

1. 使用2队列存储两组文件.请注意,应以相同的方式对两组进行排序.
2. 分别进行一些预处理dequeue.
3. 将两个预处理的张量组合成一个列表并将列表传递给 shuffle_batch

代码在这里:

base_names = ['file1', 'file2']
base_tensor = tf.convert_to_tensor(base_names)
image_name_queue = tf.train.string_input_producer(
  tensor + '.png',
  shuffle=False # Note: must set shuffle to False
)
label_queue = tf.train.string_input_producer(
  tensor + '.lable.txt',
  shuffle=False # Note: must set shuffle to False
)

# use reader to read file
image_reader = tf.WholeFileReader()
image_key, image_raw = image_reader.read(image_name_queue)
image = tf.image.decode_png(image_raw)
label_reader = tf.WholeFileReader()
label_key, label_raw = label_reader.read(label_queue)
label = tf.image.decode_raw(label_raw)

# preprocess image
processed_image = tf.image.per_image_whitening(image)
batch = tf.train.shuffle_batch([processed_image, label], 10, 100, 100)

# print batch
queue_threads = queue_runner.start_queue_runners()
print(sess.run(batch))

第2部分:Queue,QueueRunner,Coordinator和helper函数

队列实际上是一个队列(似乎毫无意义).队列有两种方法:enqueue和dequeue.输入enqueue是Tensor(嗯,您可以将正常数据排入队列,但它将被转换为Tensor内部数据).返回值dequeue是a Tensor.所以你可以像这样建立队列管道:

q1 = data_flow_ops.FIFOQueue(32, tf.int)
q2 = data_flow_ops.FIFOQueue(32, tf.int)
enq1 = q1.enqueue([1,2,3,4,5])
v1 = q1.dequeue()
enq2 = q2.enqueue(v1)

在TensorFlow中使用队列的好处是异步加载数据,这将提高性能并节省内存.上面的代码不可运行,因为没有运行这些操作的线程.QueueRunner旨在描述如何enqueue并行数据.所以初始化QueueRunner的参数是一个enqueue操作(输出enqueue).

设置完所有QueueRunners后,必须启动所有线程.一种方法是在创建它们时启动它们:

enqueue_threads = qr.create_threads(sess, coord=coord, start=True)

或者,您可以在完成所有设置工作后启动所有线程:

# add queue runner
queue_runner.add_queue_runner(queue_runner.QueueRunner(q, [enq]))

# start all queue runners
queue_threads = queue_runner.start_queue_runners()

当所有线程都启动时,您必须决定何时退出.协调员在这里这样做.Coordinator就像所有正在运行的线程之间的共享标志.如果其中一个完成或遇到错误,它将调用coord.request_stop(),然后True调用时将获得所有线程coord.should_stop().所以使用的模式Coordinator是:

coord = tf.train.Coordinator()

for step in range(1000000):
  if coord.should_stop():
    break
  print(sess.run(print_op))

coord.request_stop()
coord.join(enqueue_threads)