# Path: tf_slim/data/prefetch_queue.py
# def prefetch_queue(tensors,
#                    capacity=8,
#                    num_threads=1,
#                    dynamic_pad=False,
#                    shared_name=None,
#                    name=None):
#   """Creates a queue to prefetch tensors from `tensors`.
# 
#   A queue runner for enqueuing tensors into the prefetch_queue is automatically
#   added to the TF QueueRunners collection.
# 
#   Example:
#   This is for example useful to pre-assemble input batches read with
#   `tf.compat.v1.train.batch()` and enqueue the pre-assembled batches.  Ops that
#   dequeue
#   from the pre-assembled queue will not pay the cost of assembling the batch.
# 
#   images, labels = tf.compat.v1.train.batch([image, label], batch_size=32,
#   num_threads=4)
#   batch_queue = prefetch_queue([images, labels])
#   images, labels = batch_queue.dequeue()
#   logits = Net(images)
#   loss = Loss(logits, labels)
# 
#   Args:
#     tensors: A list or dictionary of `Tensors` to enqueue in the buffer.
#     capacity: An integer. The maximum number of elements in the queue.
#     num_threads: An integer.  Number of threads running the enqueue op.
#     dynamic_pad: Boolean.  Whether to allow variable dimensions in input shapes.
#     shared_name: (optional). If set, this queue will be shared under the given
#       name across multiple sessions.
#     name: (Optional) A name for the operations.
# 
#   Returns:
#     A queue from which you can dequeue tensors with the same type and shape
#     as `tensors`.
#   """
#   if isinstance(tensors, dict):
#     # Need to wrap the keys and values in list() since Python3 returns views.
#     # We sort the keys so the order is consistent across runs.
#     names = list(sorted(tensors.keys()))
#     tensor_list = list([tensors[n] for n in names])
#   else:
#     names = None
#     tensor_list = tensors
# 
#   with ops.name_scope(name, "prefetch_queue", tensor_list) as name:
#     dtypes = [t.dtype for t in tensor_list]
#     shapes = [t.get_shape() for t in tensor_list]
#     queue = _which_queue(dynamic_pad)(
#         capacity=capacity,
#         dtypes=dtypes,
#         shapes=shapes,
#         names=names,
#         shared_name=shared_name)
#     enqueue_op = queue.enqueue(tensors)
#     queue_runner.add_queue_runner(
#         queue_runner.QueueRunner(queue, [enqueue_op] * num_threads))
#     summary.scalar(
#         "fraction_of_%d_full" % capacity,
#         math_ops.cast(queue.size(), _dtypes.float32) * (1. / capacity))
#     return queue


import numpy as np
import tensorflow.compat.v1 as tf
from tf_slim.data import prefetch_queue
from tensorflow.python.framework import constant_op
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import errors_impl
from tensorflow.python.framework import ops
from tensorflow.python.ops import array_ops
from tensorflow.python.ops import data_flow_ops
from tensorflow.python.ops import random_ops
from tensorflow.python.ops import variables
from tensorflow.python.platform import test
from tensorflow.python.training import input as input_lib
from tensorflow.python.training import queue_runner_impl
from __future__ import print_function

# pylint:disable=g-direct-tensorflow-import
# pylint:enable=g-direct-tensorflow-import


def setUpModule():
  tf.disable_eager_execution()


class PrefetchQueueTest(test.TestCase):

  def testOneThread(self):
    with self.cached_session() as sess:
      batch_size = 10
      image_size = 32
      num_batches = 5

      zero64 = constant_op.constant(0, dtype=dtypes.int64)

      examples = variables.Variable(zero64)
      counter = examples.count_up_to(num_batches * batch_size)
      image = random_ops.random_normal(
          [image_size, image_size, 3], dtype=dtypes.float32, name='images')
      label = random_ops.random_uniform(
          [1], 0, 10, dtype=dtypes.int32, name='labels')

      batches = input_lib.batch(
          [counter, image, label], batch_size=batch_size, num_threads=1)


# Path: tf_slim/data/prefetch_queue.py
# def prefetch_queue(tensors,
#                    capacity=8,
#                    num_threads=1,
#                    dynamic_pad=False,
#                    shared_name=None,
#                    name=None):
#   """Creates a queue to prefetch tensors from `tensors`.
# 
#   A queue runner for enqueuing tensors into the prefetch_queue is automatically
#   added to the TF QueueRunners collection.
# 
#   Example:
#   This is for example useful to pre-assemble input batches read with
#   `tf.compat.v1.train.batch()` and enqueue the pre-assembled batches.  Ops that
#   dequeue
#   from the pre-assembled queue will not pay the cost of assembling the batch.
# 
#   images, labels = tf.compat.v1.train.batch([image, label], batch_size=32,
#   num_threads=4)
#   batch_queue = prefetch_queue([images, labels])
#   images, labels = batch_queue.dequeue()
#   logits = Net(images)
#   loss = Loss(logits, labels)
# 
#   Args:
#     tensors: A list or dictionary of `Tensors` to enqueue in the buffer.
#     capacity: An integer. The maximum number of elements in the queue.
#     num_threads: An integer.  Number of threads running the enqueue op.
#     dynamic_pad: Boolean.  Whether to allow variable dimensions in input shapes.
#     shared_name: (optional). If set, this queue will be shared under the given
#       name across multiple sessions.
#     name: (Optional) A name for the operations.
# 
#   Returns:
#     A queue from which you can dequeue tensors with the same type and shape
#     as `tensors`.
#   """
#   if isinstance(tensors, dict):
#     # Need to wrap the keys and values in list() since Python3 returns views.
#     # We sort the keys so the order is consistent across runs.
#     names = list(sorted(tensors.keys()))
#     tensor_list = list([tensors[n] for n in names])
#   else:
#     names = None
#     tensor_list = tensors
# 
#   with ops.name_scope(name, "prefetch_queue", tensor_list) as name:
#     dtypes = [t.dtype for t in tensor_list]
#     shapes = [t.get_shape() for t in tensor_list]
#     queue = _which_queue(dynamic_pad)(
#         capacity=capacity,
#         dtypes=dtypes,
#         shapes=shapes,
#         names=names,
#         shared_name=shared_name)
#     enqueue_op = queue.enqueue(tensors)
#     queue_runner.add_queue_runner(
#         queue_runner.QueueRunner(queue, [enqueue_op] * num_threads))
#     summary.scalar(
#         "fraction_of_%d_full" % capacity,
#         math_ops.cast(queue.size(), _dtypes.float32) * (1. / capacity))
#     return queue


# Path: tf_slim/data/prefetch_queue_test.py
import numpy as np
import tensorflow.compat.v1 as tf
from tf_slim.data import prefetch_queue
from tensorflow.python.framework import constant_op
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import errors_impl
from tensorflow.python.framework import ops
from tensorflow.python.ops import array_ops
from tensorflow.python.ops import data_flow_ops
from tensorflow.python.ops import random_ops
from tensorflow.python.ops import variables
from tensorflow.python.platform import test
from tensorflow.python.training import input as input_lib
from tensorflow.python.training import queue_runner_impl

from __future__ import print_function

# pylint:disable=g-direct-tensorflow-import
# pylint:enable=g-direct-tensorflow-import


def setUpModule():
  tf.disable_eager_execution()


class PrefetchQueueTest(test.TestCase):

  def testOneThread(self):
    with self.cached_session() as sess:
      batch_size = 10
      image_size = 32
      num_batches = 5

      zero64 = constant_op.constant(0, dtype=dtypes.int64)

      examples = variables.Variable(zero64)
      counter = examples.count_up_to(num_batches * batch_size)
      image = random_ops.random_normal(
          [image_size, image_size, 3], dtype=dtypes.float32, name='images')
      label = random_ops.random_uniform(
          [1], 0, 10, dtype=dtypes.int32, name='labels')

      batches = input_lib.batch(
          [counter, image, label], batch_size=batch_size, num_threads=1)


      batches = prefetch_queue.prefetch_queue(batches).dequeue()