text: #   Copyright (c) 2018 PaddlePaddle 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.
  
  from __future__ import print_function
  
  from . import core
  import numpy
  import os
  import six
  from six.moves import zip, range, xrange
  import multiprocessing
  
  from .framework import Variable, default_main_program, _current_expected_place
  from .framework import _cpu_num, _cuda_ids
  __all__ = ['DataFeeder']
  
  
  def convert_dtype(dtype):
      if dtype == core.VarDesc.VarType.FP32:
          return 'float32'
      elif dtype == core.VarDesc.VarType.INT64:
          return 'int64'
      elif dtype == core.VarDesc.VarType.FP64:
          return 'float64'
      elif dtype == core.VarDesc.VarType.FP16:
          return 'float16'
      elif dtype == core.VarDesc.VarType.INT32:
          return 'int32'
      elif dtype == core.VarDesc.VarType.UINT8:
          return 'uint8'
      else:
          raise ValueError("dtype must be any of [int32, float32, int64, "
                           "float64, uint8]")
  
  
  class DataToLoDTensorConverter(object):
      def __init__(self, place, lod_level, shape, dtype):
          self.place = place
          self.lod_level = lod_level
          self.shape = shape
          negtive_count = 0
          for s in self.shape:
              if s < 0:
                  negtive_count += 1
              if negtive_count > 1:
                  self.shape = None
                  break
          self.dtype = convert_dtype(dtype)
          self._reset()
  
      def _reset(self):
          self.data = []
          self.lod = [[] for _ in six.moves.range(self.lod_level)]
  
      def feed(self, data):
          self._feed_impl_(data, self.lod, self.lod_level)
  
      def _feed_impl_(self, data, lod, lod_level):
          if lod_level == 0:
              self.data.append(data)
          else:
              lod[0].append(len(data))
              for each_data in data:
                  self._feed_impl_(each_data, lod[1:], lod_level - 1)
  
      def _check_shape(self, shape):
          for s1, s2 in zip(self.shape, shape):
              if s1 != s2 and s1 >= 0 and s2 >= 0:
                  raise ValueError(
                      "Shape not match. What is defined in data layer is {}, but receive {}".
                      format(self.shape, shape))
  
      def done(self):
          arr = numpy.array(self.data, dtype=self.dtype)
          if self.shape:
              if len(arr.shape) != len(self.shape):
                  try:
                      arr = arr.reshape(self.shape)
                  except ValueError:
                      raise ValueError(
                          "Reshape error. What is defined in data layer is {}, but receive {}"
                          .format(self.shape, arr.shape))
          t = core.LoDTensor()
          t.set(arr, self.place)
          if self.lod_level > 0:
              t.set_recursive_sequence_lengths(self.lod)
          self._reset()
          return t
  
  
  class BatchedTensorProvider(object):
      def __init__(self, feed_list, place, batch_size, generator, drop_last):
          self.place = place
          self.batch_size = batch_size
          self.generator = generator
          self.converters = []
          self.drop_last = drop_last
  
          for var in feed_list:
              assert var.lod_level == 0, "lod_level must be 0"
              self.converters.append(
                  DataToLoDTensorConverter(
                      place=self.place,
                      lod_level=0,
                      shape=var.shape,
                      dtype=var.dtype))
  
      def _done(self):
          return [c.done() for c in self.converters]
  
      def __call__(self):
          idx = 0
          for each_sample in self.generator():
              for each_slot, each_converter in six.moves.zip(each_sample,
                                                             self.converters):
                  each_converter.data.append(each_slot)
  
              idx += 1
              if idx == self.batch_size:
                  idx = 0
                  yield self._done()
  
          if not self.drop_last and idx > 0:
              yield self._done()
          else:
              [c._reset() for c in self.converters]
  
  
  class DataFeeder(object):
      """
      DataFeeder converts the data that returned by a reader into a data
      structure that can feed into Executor and ParallelExecutor. The reader
      usually returns a list of mini-batch data entries. Each data entry in
      the list is one sample. Each sample is a list or a tuple with one
      feature or multiple features.
  
      The simple usage shows below:
  
      ..  code-block:: python
  
          import paddle.fluid as fluid
          place = fluid.CPUPlace()
          img = fluid.layers.data(name='image', shape=[1, 28, 28])
          label = fluid.layers.data(name='label', shape=[1], dtype='int64')
          feeder = fluid.DataFeeder([img, label], fluid.CPUPlace())
          result = feeder.feed([([0] * 784, [9]), ([1] * 784, [1])])
  
  
      If you want to feed data into GPU side separately in advance when you
      use multi-GPU to train a model, you can use `decorate_reader` function.
  
      ..  code-block:: python
  
          import paddle
          import paddle.fluid as fluid
          
          place=fluid.CUDAPlace(0)
          data = fluid.layers.data(name='data', shape=[3, 224, 224], dtype='float32')
          label = fluid.layers.data(name='label', shape=[1], dtype='int64')
          
          feeder = fluid.DataFeeder(place=place, feed_list=[data, label])
          reader = feeder.decorate_reader(
                  paddle.batch(paddle.dataset.flowers.train(), batch_size=16), multi_devices=False)
  
      Args:
          feed_list(list): The Variables or Variables'name that will
              feed into model.
          place(Place): place indicates feed data into CPU or GPU, if you want to
              feed data into GPU, please using `fluid.CUDAPlace(i)` (`i` represents
              the GPU id), or if you want to feed data into CPU, please using
              `fluid.CPUPlace()`.
          program(Program): The Program that will feed data into, if program
              is None, it will use default_main_program(). Default None.
  
      Raises:
          ValueError: If some Variable is not in this Program.
  
      Examples:
          ..  code-block:: python
  
  
              import numpy as np
              import paddle
              import paddle.fluid as fluid
              
              place = fluid.CPUPlace()
              
              def reader():
                  yield [np.random.random([4]).astype('float32'), np.random.random([3]).astype('float32')],
              
              main_program = fluid.Program()
              startup_program = fluid.Program()
              
              with fluid.program_guard(main_program, startup_program):
                  data_1 = fluid.layers.data(name='data_1', shape=[1, 2, 2])
                  data_2 = fluid.layers.data(name='data_2', shape=[1, 1, 3])
                  out = fluid.layers.fc(input=[data_1, data_2], size=2)
                  # ...
              
              feeder = fluid.DataFeeder([data_1, data_2], place)
                          
              exe = fluid.Executor(place)
              exe.run(startup_program)
              for data in reader():
                  outs = exe.run(program=main_program,
                                 feed=feeder.feed(data),
                                 fetch_list=[out])
  
      """
  
      def __init__(self, feed_list, place, program=None):
          self.feed_dtypes = []
          self.feed_names = []
          self.feed_shapes = []
          self.feed_lod_level = []
          if program is None:
              program = default_main_program()
          for each_var in feed_list:
              if isinstance(each_var, six.string_types):
                  each_var = program.block(0).var(each_var)
              if not isinstance(each_var, Variable):
                  raise TypeError("Feed list should contain a list of variable")
              self.feed_dtypes.append(each_var.dtype)
              self.feed_names.append(each_var.name)
              self.feed_lod_level.append(each_var.lod_level)
              self.feed_shapes.append(each_var.shape)
  
          self.place = place
  
      def feed(self, iterable):
          """
          According to feed_list and iterable, converters the input into
          a data structure that can feed into Executor and ParallelExecutor.
  
          Args:
              iterable(list|tuple): the input data.
  
          Returns:
              dict: the result of conversion.
  
          Examples:
              ..  code-block:: python
  
                  import numpy.random as random
                  import paddle.fluid as fluid
                  
                  def reader(limit=5):
                      for i in range(limit):
                          yield random.random([784]).astype('float32'), random.random([1]).astype('int64'), random.random([256]).astype('float32')
                  
                  data_1 = fluid.layers.data(name='data_1', shape=[1, 28, 28])
                  data_2 = fluid.layers.data(name='data_2', shape=[1], dtype='int64')
                  data_3 = fluid.layers.data(name='data_3', shape=[16, 16], dtype='float32')
                  feeder = fluid.DataFeeder(['data_1','data_2', 'data_3'], fluid.CPUPlace())
                  
                  result = feeder.feed(reader()) 
          """
          converter = []
          for lod_level, shape, dtype in six.moves.zip(
                  self.feed_lod_level, self.feed_shapes, self.feed_dtypes):
              converter.append(
                  DataToLoDTensorConverter(
                      place=self.place,
                      lod_level=lod_level,
                      shape=shape,
                      dtype=dtype))
  
          for each_sample in iterable:
              assert len(each_sample) == len(converter), (
                  "The number of fields in data (%s) does not match " +
                  "len(feed_list) (%s)") % (len(each_sample), len(converter))
              for each_converter, each_slot in six.moves.zip(converter,
                                                             each_sample):
                  each_converter.feed(each_slot)
          ret_dict = {}
          for each_name, each_converter in six.moves.zip(self.feed_names,
                                                         converter):
              ret_dict[each_name] = each_converter.done()
          return ret_dict
  
      def feed_parallel(self, iterable, num_places=None):
          """
          Takes multiple mini-batches. Each mini-batch will be feed on each
          device in advance.
  
          Args:
              iterable(list|tuple): the input data.
              num_places(int): the number of devices. Default None.
  
          Returns:
              dict: the result of conversion.
  
          Notes:
              The number of devices and number of mini-batches must be same.
  
          Examples:
              ..  code-block:: python
  
                  import numpy.random as random
                  import paddle.fluid as fluid
                  
                  def reader(limit=10):
                      for i in range(limit):
                          yield [random.random([784]).astype('float32'), random.random([1]).astype('float32')],
                  
                  x = fluid.layers.data(name='x', shape=[1, 28, 28])
                  y = fluid.layers.data(name='y', shape=[1], dtype='float32')
                  
                  fluid.layers.elementwise_add(x, y)
                  
                  feeder = fluid.DataFeeder(['x','y'], fluid.CPUPlace())
                  place_num = 2 
                  places = [fluid.CPUPlace() for x in range(place_num)]
                  data = []
                  exe = fluid.Executor(fluid.CPUPlace())
                  exe.run(fluid.default_startup_program())
                  program = fluid.CompiledProgram(fluid.default_main_program()).with_data_parallel(places=places)
                  for item in reader():
                      data.append(item)
                      if place_num == len(data):
                          exe.run(program=program, feed=list(feeder.feed_parallel(data, place_num)), fetch_list=[])
                          data = []
          """
          if isinstance(self.place, core.CUDAPlace):
              places = [
                  core.CUDAPlace(i)
                  for i in six.moves.xrange(
                      self._get_number_of_places_(num_places))
              ]
          else:
              places = [
                  core.CPUPlace()
                  for _ in six.moves.xrange(
                      self._get_number_of_places_(num_places))
              ]
  
          if len(iterable) != len(places):
              raise ValueError("feed_parallel takes multiple mini-batches. Each "
                               "mini-batch will be feed on each device. The "
                               "number of devices and number of mini-batches "
                               "must be same.")
  
          place = self.place
          for p, batch in six.moves.zip(places, iterable):
              self.place = p
              yield self.feed(batch)
          self.place = place
  
      def _get_number_of_places_(self, num_places):
          if num_places is not None:
              return int(num_places)
          elif isinstance(self.place, core.CUDAPlace):
              return len(_cuda_ids())
          else:
              return _cpu_num()
  
      def decorate_reader(self,
                          reader,
                          multi_devices,
                          num_places=None,
                          drop_last=True):
          """
          Converter the input data into a data that returned by reader into
          multiple mini-batches. Each mini-batch will be feed on each device.
  
          Args:
              reader(function): the reader is the function which can generate data.
              multi_devices(bool): whether to use multiple devices or not.
              num_places(int): if multi_devices is True, you can specify the number
                  of GPU to use, if multi_devices is None, the function will use all the
                  GPU of the current machine. Default None.
              drop_last(bool): whether to drop the last batch if the
                  size of the last batch is less than batch_size. Default True.
  
          Returns:
              dict: the result of conversion.
  
          Raises:
              ValueError: If drop_last is False and the data batch cannot fit for devices.
  
          Examples:
              ..  code-block:: python
  
                  import numpy.random as random
                  import paddle
                  import paddle.fluid as fluid
                  
                  def reader(limit=5):
                      for i in range(limit):
                          yield (random.random([784]).astype('float32'), random.random([1]).astype('int64')),
                  
                  place=fluid.CPUPlace()
                  data = fluid.layers.data(name='data', shape=[1, 28, 28], dtype='float32')
                  label = fluid.layers.data(name='label', shape=[1], dtype='int64')
                  
                  feeder = fluid.DataFeeder(place=place, feed_list=[data, label])
                  reader = feeder.decorate_reader(reader, multi_devices=False)
                  
                  exe = fluid.Executor(place)
                  exe.run(fluid.default_startup_program())
                  for data in reader():
                      exe.run(feed=data)
          """
  
          def __reader_creator__():
              if not multi_devices:
                  for item in reader():
                      yield self.feed(item)
              else:
                  num = self._get_number_of_places_(num_places)
                  item = []
                  for batch in reader():
                      item.append(batch)
                      if len(item) == num:
                          yield list(self.feed_parallel(item, num))
                          item = []
                  if not drop_last and len(item) != 0:
                      raise ValueError(
                          "The data batch which cannot fit for devices will be "
                          "dropped is not implementation. Other strategies are "
                          "not implemented")
  
          return __reader_creator__
  
  
  class NumpyToLoDTensorConverter(object):
      def __init__(self, place):
          self.place = place
          self.data = []
          self._reset()
  
      def _reset(self):
          self.data = []
  
      def feed(self, data):
          self.data.append(data)
  
      def done(self):
          arr = numpy.array(self.data)
          t = core.LoDTensor()
          t.set(arr, self.place)
          self._reset()
          return t
  
  
  class ListTensorProvider(object):
      def __init__(self, generator, places):
          self.generator = generator
          self.converters = []
          self.places = []
          if places:
              if not isinstance(places, (list, tuple)):
                  places = [places]
              assert len(
                  places) == 1, "dygraph mode CAN NOT specify multiple places."
              for place in places:
                  if isinstance(place, (core.CUDAPlace, core.CPUPlace)):
                      self.places.append(place)
                  else:
                      raise ValueError(
                          "Please specify a valid place values such as core.CPUPlace or core.CUDAPlace"
                      )
          if len(self.places) == 0:
              self.places.append(_current_expected_place())
  
      def _readData(self, iterable, places):
          for place, each_sample in six.moves.zip(places, iterable):
              for item in each_sample:
                  if len(self.converters) < len(item):
                      for i in item:
                          self.converters.append(NumpyToLoDTensorConverter(place))
                  for each_converter, each_slot in six.moves.zip(self.converters,
                                                                 item):
                      each_converter.feed(each_slot)
              yield [c.done() for c in self.converters]
  
      def __call__(self):
          item = []
          for batch in self.generator():
              item.append(batch)
              if len(item) == len(self.places):
                  yield list(self._readData(item, self.places))
                  item = []
  