# Copyright 2016 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 for sequence data preprocessing utils."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

from math import ceil

import numpy as np

from tensorflow.python import keras
from tensorflow.python.platform import test


class TestSequence(test.TestCase):

  def test_pad_sequences(self):
    a = [[1], [1, 2], [1, 2, 3]]

    # test padding
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, padding='pre')
    self.assertAllClose(b, [[0, 0, 1], [0, 1, 2], [1, 2, 3]])
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, padding='post')
    self.assertAllClose(b, [[1, 0, 0], [1, 2, 0], [1, 2, 3]])

    # test truncating
    b = keras.preprocessing.sequence.pad_sequences(
        a, maxlen=2, truncating='pre')
    self.assertAllClose(b, [[0, 1], [1, 2], [2, 3]])
    b = keras.preprocessing.sequence.pad_sequences(
        a, maxlen=2, truncating='post')
    self.assertAllClose(b, [[0, 1], [1, 2], [1, 2]])

    # test value
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, value=1)
    self.assertAllClose(b, [[1, 1, 1], [1, 1, 2], [1, 2, 3]])

  def test_pad_sequences_vector(self):
    a = [[[1, 1]], [[2, 1], [2, 2]], [[3, 1], [3, 2], [3, 3]]]

    # test padding
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, padding='pre')
    self.assertAllClose(b, [[[0, 0], [0, 0], [1, 1]], [[0, 0], [2, 1], [2, 2]],
                            [[3, 1], [3, 2], [3, 3]]])
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, padding='post')
    self.assertAllClose(b, [[[1, 1], [0, 0], [0, 0]], [[2, 1], [2, 2], [0, 0]],
                            [[3, 1], [3, 2], [3, 3]]])

    # test truncating
    b = keras.preprocessing.sequence.pad_sequences(
        a, maxlen=2, truncating='pre')
    self.assertAllClose(b, [[[0, 0], [1, 1]], [[2, 1], [2, 2]], [[3, 2], [3,
                                                                          3]]])

    b = keras.preprocessing.sequence.pad_sequences(
        a, maxlen=2, truncating='post')
    self.assertAllClose(b, [[[0, 0], [1, 1]], [[2, 1], [2, 2]], [[3, 1], [3,
                                                                          2]]])

    # test value
    b = keras.preprocessing.sequence.pad_sequences(a, maxlen=3, value=1)
    self.assertAllClose(b, [[[1, 1], [1, 1], [1, 1]], [[1, 1], [2, 1], [2, 2]],
                            [[3, 1], [3, 2], [3, 3]]])

  def test_make_sampling_table(self):
    a = keras.preprocessing.sequence.make_sampling_table(3)
    self.assertAllClose(
        a, np.asarray([0.00315225, 0.00315225, 0.00547597]), rtol=.1)

  def test_skipgrams(self):
    # test with no window size and binary labels
    couples, labels = keras.preprocessing.sequence.skipgrams(
        np.arange(3), vocabulary_size=3)
    for couple in couples:
      self.assertIn(couple[0], [0, 1, 2])
      self.assertIn(couple[1], [0, 1, 2])

    # test window size and categorical labels
    couples, labels = keras.preprocessing.sequence.skipgrams(
        np.arange(5), vocabulary_size=5, window_size=1, categorical=True)
    for couple in couples:
      self.assertLessEqual(couple[0] - couple[1], 3)
    for l in labels:
      self.assertEqual(len(l), 2)

  def test_remove_long_seq(self):
    a = [[[1, 1]], [[2, 1], [2, 2]], [[3, 1], [3, 2], [3, 3]]]

    new_seq, new_label = keras.preprocessing.sequence._remove_long_seq(
        maxlen=3, seq=a, label=['a', 'b', ['c', 'd']])
    self.assertEqual(new_seq, [[[1, 1]], [[2, 1], [2, 2]]])
    self.assertEqual(new_label, ['a', 'b'])

  def test_TimeseriesGenerator(self):
    data = np.array([[i] for i in range(50)])
    targets = np.array([[i] for i in range(50)])

    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data, targets, length=10, sampling_rate=2, batch_size=2)
    self.assertEqual(len(data_gen), 20)
    self.assertAllClose(data_gen[0][0],
                        np.array([[[0], [2], [4], [6], [8]], [[1], [3], [5],
                                                              [7], [9]]]))
    self.assertAllClose(data_gen[0][1], np.array([[10], [11]]))
    self.assertAllClose(data_gen[1][0],
                        np.array([[[2], [4], [6], [8], [10]], [[3], [5], [7],
                                                               [9], [11]]]))
    self.assertAllClose(data_gen[1][1], np.array([[12], [13]]))

    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data, targets, length=10, sampling_rate=2, reverse=True, batch_size=2)
    self.assertEqual(len(data_gen), 20)
    self.assertAllClose(data_gen[0][0],
                        np.array([[[8], [6], [4], [2], [0]], [[9], [7], [5],
                                                              [3], [1]]]))
    self.assertAllClose(data_gen[0][1], np.array([[10], [11]]))

    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data, targets, length=10, sampling_rate=2, shuffle=True, batch_size=1)
    batch = data_gen[0]
    r = batch[1][0][0]
    self.assertAllClose(batch[0],
                        np.array([[[r - 10], [r - 8], [r - 6], [r - 4],
                                   [r - 2]]]))
    self.assertAllClose(batch[1], np.array([
        [r],
    ]))

    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data, targets, length=10, sampling_rate=2, stride=2, batch_size=2)
    self.assertEqual(len(data_gen), 10)
    self.assertAllClose(data_gen[1][0],
                        np.array([[[4], [6], [8], [10], [12]], [[6], [8], [10],
                                                                [12], [14]]]))
    self.assertAllClose(data_gen[1][1], np.array([[14], [16]]))

    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data,
        targets,
        length=10,
        sampling_rate=2,
        start_index=10,
        end_index=30,
        batch_size=2)
    self.assertEqual(len(data_gen), 6)
    self.assertAllClose(data_gen[0][0],
                        np.array([[[10], [12], [14], [16], [18]],
                                  [[11], [13], [15], [17], [19]]]))
    self.assertAllClose(data_gen[0][1], np.array([[20], [21]]))

    data = np.array([np.random.random_sample((1, 2, 3, 4)) for i in range(50)])
    targets = np.array([np.random.random_sample((3, 2, 1)) for i in range(50)])
    data_gen = keras.preprocessing.sequence.TimeseriesGenerator(
        data,
        targets,
        length=10,
        sampling_rate=2,
        start_index=10,
        end_index=30,
        batch_size=2)

    self.assertEqual(len(data_gen), 6)
    self.assertAllClose(data_gen[0][0],
                        np.array(
                            [np.array(data[10:19:2]),
                             np.array(data[11:20:2])]))
    self.assertAllClose(data_gen[0][1], np.array([targets[20], targets[21]]))

    with self.assertRaises(ValueError) as context:
      keras.preprocessing.sequence.TimeseriesGenerator(data, targets, length=50)
    error = str(context.exception)
    self.assertIn('`start_index+length=50 > end_index=49` is disallowed', error)

  def test_TimeSeriesGenerator_doesnt_miss_any_sample(self):
    x = np.array([[i] for i in range(10)])

    for length in range(3, 10):
      g = keras.preprocessing.sequence.TimeseriesGenerator(
          x, x, length=length, batch_size=1)
      expected = max(0, len(x) - length)
      actual = len(g)
      self.assertEqual(expected, actual)

      if actual > 0:
        # All elements in range(length, 10) should be used as current step
        expected = np.arange(length, 10).reshape(-1, 1)

        y = np.concatenate([g[ix][1] for ix in range(len(g))], axis=0)
        self.assertAllClose(y, expected)

    x = np.array([[i] for i in range(23)])

    strides = (1, 1, 5, 7, 3, 5, 3)
    lengths = (3, 3, 4, 3, 1, 3, 7)
    batch_sizes = (6, 6, 6, 5, 6, 6, 6)
    shuffles = (False, True, True, False, False, False, False)

    for stride, length, batch_size, shuffle in zip(strides, lengths,
                                                   batch_sizes, shuffles):
      g = keras.preprocessing.sequence.TimeseriesGenerator(
          x,
          x,
          length=length,
          sampling_rate=1,
          stride=stride,
          start_index=0,
          end_index=None,
          shuffle=shuffle,
          reverse=False,
          batch_size=batch_size)
      if shuffle:
        # all batches have the same size when shuffle is True.
        expected_sequences = ceil(
            (23 - length) / float(batch_size * stride)) * batch_size
      else:
        # last batch will be different if `(samples - length) / stride`
        # is not a multiple of `batch_size`.
        expected_sequences = ceil((23 - length) / float(stride))

      expected_batches = ceil(expected_sequences / float(batch_size))

      y = [g[ix][1] for ix in range(len(g))]

      actual_sequences = sum(len(iy) for iy in y)
      actual_batches = len(y)

      self.assertEqual(expected_sequences, actual_sequences)
      self.assertEqual(expected_batches, actual_batches)


if __name__ == '__main__':
  test.main()
