# Path: tf_slim/queues.py
# class NestedQueueRunnerError(Exception):
# def QueueRunners(session):
# 
# Path: tf_slim/data/parallel_reader.py
# class ParallelReader(io_ops.ReaderBase):
#   def __init__(self,
#                reader_class,
#                common_queue,
#                num_readers=4,
#                reader_kwargs=None):
#   def num_readers(self):
#   def common_queue(self):
#   def read(self, queue, name=None):
#   def read_up_to(self, queue, num_records, name=None):
#   def _configure_readers_by(self, queue):
#   def num_records_produced(self, name=None):
#   def num_work_units_completed(self, name=None):
# def parallel_read(data_sources,
#                   reader_class,
#                   num_epochs=None,
#                   num_readers=4,
#                   reader_kwargs=None,
#                   shuffle=True,
#                   dtypes=None,
#                   capacity=256,
#                   min_after_dequeue=128,
#                   seed=None,
#                   scope=None):
# def single_pass_read(data_sources, reader_class, reader_kwargs=None,
#                      scope=None):
# def get_data_files(data_sources):
# 
# Path: tf_slim/data/test_utils.py
# def _encoded_int64_feature(ndarray):
# def _encoded_bytes_feature(tf_encoded):
#   def string_to_bytes(value):
# def _string_feature(value):
# def _encoder(image, image_format):
# def generate_image(image_shape, image_format='jpeg', label=0):
# def create_tfrecord_files(output_dir, num_files=3, num_records_per_file=10):


import tempfile
import tensorflow.compat.v1 as tf
from tf_slim import queues
from tf_slim.data import parallel_reader
from tf_slim.data import test_utils
from tensorflow.python.framework import dtypes as dtypes_lib
from tensorflow.python.framework import errors_impl
from tensorflow.python.framework import ops
from tensorflow.python.ops import data_flow_ops
from tensorflow.python.ops import io_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 supervisor
    with sv.prepare_or_wait_for_session() as sess:
      sv.start_queue_runners(sess)

      flowers = 0
      num_reads = 100
      for _ in range(num_reads):
        current_key, _ = sess.run([key, value])
        if 'flowers' in str(current_key):
          flowers += 1
      self.assertGreater(flowers, 0)
      self.assertEqual(flowers, num_reads)


class SinglePassReadTest(test.TestCase):

  def setUp(self):
    super(SinglePassReadTest, self).setUp()
    ops.reset_default_graph()

  def testOutOfRangeError(self):
    with self.cached_session():
      [tfrecord_path] = test_utils.create_tfrecord_files(
          tempfile.mkdtemp(), num_files=1)

    key, value = parallel_reader.single_pass_read(
        tfrecord_path, reader_class=io_ops.TFRecordReader)
    init_op = variables.local_variables_initializer()

    with self.cached_session() as sess:
      sess.run(init_op)

# Path: tf_slim/queues.py
# class NestedQueueRunnerError(Exception):
# def QueueRunners(session):
# 
# Path: tf_slim/data/parallel_reader.py
# class ParallelReader(io_ops.ReaderBase):
#   def __init__(self,
#                reader_class,
#                common_queue,
#                num_readers=4,
#                reader_kwargs=None):
#   def num_readers(self):
#   def common_queue(self):
#   def read(self, queue, name=None):
#   def read_up_to(self, queue, num_records, name=None):
#   def _configure_readers_by(self, queue):
#   def num_records_produced(self, name=None):
#   def num_work_units_completed(self, name=None):
# def parallel_read(data_sources,
#                   reader_class,
#                   num_epochs=None,
#                   num_readers=4,
#                   reader_kwargs=None,
#                   shuffle=True,
#                   dtypes=None,
#                   capacity=256,
#                   min_after_dequeue=128,
#                   seed=None,
#                   scope=None):
# def single_pass_read(data_sources, reader_class, reader_kwargs=None,
#                      scope=None):
# def get_data_files(data_sources):
# 
# Path: tf_slim/data/test_utils.py
# def _encoded_int64_feature(ndarray):
# def _encoded_bytes_feature(tf_encoded):
#   def string_to_bytes(value):
# def _string_feature(value):
# def _encoder(image, image_format):
# def generate_image(image_shape, image_format='jpeg', label=0):
# def create_tfrecord_files(output_dir, num_files=3, num_records_per_file=10):


# Path: tf_slim/data/parallel_reader_test.py
import tempfile
import tensorflow.compat.v1 as tf
from tf_slim import queues
from tf_slim.data import parallel_reader
from tf_slim.data import test_utils
from tensorflow.python.framework import dtypes as dtypes_lib
from tensorflow.python.framework import errors_impl
from tensorflow.python.framework import ops
from tensorflow.python.ops import data_flow_ops
from tensorflow.python.ops import io_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 supervisor

    with sv.prepare_or_wait_for_session() as sess:
      sv.start_queue_runners(sess)

      flowers = 0
      num_reads = 100
      for _ in range(num_reads):
        current_key, _ = sess.run([key, value])
        if 'flowers' in str(current_key):
          flowers += 1
      self.assertGreater(flowers, 0)
      self.assertEqual(flowers, num_reads)


class SinglePassReadTest(test.TestCase):

  def setUp(self):
    super(SinglePassReadTest, self).setUp()
    ops.reset_default_graph()

  def testOutOfRangeError(self):
    with self.cached_session():
      [tfrecord_path] = test_utils.create_tfrecord_files(
          tempfile.mkdtemp(), num_files=1)

    key, value = parallel_reader.single_pass_read(
        tfrecord_path, reader_class=io_ops.TFRecordReader)
    init_op = variables.local_variables_initializer()

    with self.cached_session() as sess:
      sess.run(init_op)

      with queues.QueueRunners(sess):