content: # Copyright 2015 The TensorFlow Authors. All Rights Reserved.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  # ==============================================================================
  """Tests NumpySource and PandasSource."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  import numpy as np
  
  from tensorflow.contrib.learn.python.learn.dataframe.transforms import in_memory_source
  from tensorflow.python.client import session
  from tensorflow.python.framework import ops
  from tensorflow.python.platform import test
  from tensorflow.python.training import coordinator
  from tensorflow.python.training import queue_runner_impl
  
  # pylint: disable=g-import-not-at-top
  try:
    import pandas as pd
    HAS_PANDAS = True
  except ImportError:
    HAS_PANDAS = False
  
  
  def get_rows(array, row_indices):
    rows = [array[i] for i in row_indices]
    return np.vstack(rows)
  
  
  class NumpySourceTestCase(test.TestCase):
  
    def testNumpySource(self):
      batch_size = 3
      iterations = 1000
      array = np.arange(32).reshape([16, 2])
      numpy_source = in_memory_source.NumpySource(array, batch_size=batch_size)
      index_column = numpy_source().index
      value_column = numpy_source().value
      cache = {}
      with ops.Graph().as_default():
        value_tensor = value_column.build(cache)
        index_tensor = index_column.build(cache)
        with session.Session() as sess:
          coord = coordinator.Coordinator()
          threads = queue_runner_impl.start_queue_runners(sess=sess, coord=coord)
          for i in range(iterations):
            expected_index = [
                j % array.shape[0]
                for j in range(batch_size * i, batch_size * (i + 1))
            ]
            expected_value = get_rows(array, expected_index)
            actual_index, actual_value = sess.run([index_tensor, value_tensor])
            np.testing.assert_array_equal(expected_index, actual_index)
            np.testing.assert_array_equal(expected_value, actual_value)
          coord.request_stop()
          coord.join(threads)
  
  
  class PandasSourceTestCase(test.TestCase):
  
    def testPandasFeeding(self):
      if not HAS_PANDAS:
        return
      batch_size = 3
      iterations = 1000
      index = np.arange(100, 132)
      a = np.arange(32)
      b = np.arange(32, 64)
      dataframe = pd.DataFrame({"a": a, "b": b}, index=index)
      pandas_source = in_memory_source.PandasSource(
          dataframe, batch_size=batch_size)
      pandas_columns = pandas_source()
      cache = {}
      with ops.Graph().as_default():
        pandas_tensors = [col.build(cache) for col in pandas_columns]
        with session.Session() as sess:
          coord = coordinator.Coordinator()
          threads = queue_runner_impl.start_queue_runners(sess=sess, coord=coord)
          for i in range(iterations):
            indices = [
                j % dataframe.shape[0]
                for j in range(batch_size * i, batch_size * (i + 1))
            ]
            expected_df_indices = dataframe.index[indices]
            expected_rows = dataframe.iloc[indices]
            actual_value = sess.run(pandas_tensors)
            np.testing.assert_array_equal(expected_df_indices, actual_value[0])
            for col_num, col in enumerate(dataframe.columns):
              np.testing.assert_array_equal(expected_rows[col].values,
                                            actual_value[col_num + 1])
          coord.request_stop()
          coord.join(threads)
  
  
  if __name__ == "__main__":
    test.main()
  