code: # SPDX-License-Identifier: Apache-2.0
  
  
  """Unit Tests for Tensorflow shape inference."""
  
  from __future__ import division
  from __future__ import print_function
  from __future__ import unicode_literals
  
  import os
  import numpy as np
  import tensorflow as tf
  
  from tensorflow.python.ops import variables as variables_lib
  from tensorflow.python.ops import init_ops
  
  from backend_test_base import Tf2OnnxBackendTestBase
  from common import *  # pylint: disable=wildcard-import, unused-wildcard-import
  from tf2onnxnightly import utils
  from tf2onnxnightly.tf_utils import get_tf_tensor_shape
  from tf2onnxnightly.shape_inference import infer_shape_for_graph
  from tf2onnxnightly.tf_loader import tf_reset_default_graph, tf_session, tf_placeholder, tf_optimize
  
  # pylint: disable=missing-docstring
  
  
  class TFShapeInferenceTests(Tf2OnnxBackendTestBase):
      def _run_test_case(self, input_names_with_port, output_names_with_port):
          try:
              tf.compat.v1.disable_eager_execution()
          except:  # pylint: disable=bare-except
              pass
          graph_def = None
          with tf_session() as sess:
              # freeze graph
              origin_graph = sess.graph
              variables_lib.global_variables_initializer().run()
              output_name_without_port = [n.split(':')[0] for n in output_names_with_port]
              graph_def = tf.graph_util.convert_variables_to_constants(
                  sess, sess.graph_def,
                  output_name_without_port
              )
  
          tf_reset_default_graph()
          tf.import_graph_def(graph_def, name='')
  
          # optimize graph
          graph_def = tf_optimize(input_names_with_port, output_names_with_port, sess.graph_def, True)
  
          with tf_session() as sess:
              if self.config.is_debug_mode:
                  if not os.path.exists(self.test_data_directory):
                      os.makedirs(self.test_data_directory)
                  model_path = os.path.join(self.test_data_directory, self._testMethodName + "_after_tf_optimize.pb")
                  utils.save_protobuf(model_path, graph_def)
                  self.logger.debug("created file  %s", model_path)
  
          tf_reset_default_graph()
          tf.import_graph_def(graph_def, name='')
  
          with tf_session() as sess:
              inferred_graph = infer_shape_for_graph(sess.graph)
              # compare each operation
              for op in origin_graph.get_operations():
                  inferred_op = None
                  try:
                      inferred_op = inferred_graph.get_operation_by_name(op.name)
                  except KeyError:
                      continue
                  self._compare_shape_for_op(op, inferred_op)
  
      def _compare_shape_for_op(self, op1, op2):
          """Align outputs of op2 to op1."""
          for out1, out2 in zip(op1.outputs, op2.outputs):
              expected_shape = get_tf_tensor_shape(out1)
              if out1 is not None:
                  actual_shape = get_tf_tensor_shape(out2)
                  self.assertTrue(utils.are_shapes_compatible(expected_shape, actual_shape))
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_while_loop_with_ta_read_and_write(self):
          i = tf_placeholder(tf.int32, (), name="input_1")
          inputs = tf_placeholder(tf.float32, (10,), name="input_2")
  
          inputs_2 = tf.identity(inputs)
          input_ta = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True).unstack(inputs_2)
          output_ta = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  
          c = lambda i, *_: tf.logical_and(tf.less(i, 10), i >= 0)
  
          def b(i, out_ta):
              new_i = tf.add(i, 1)
              x = input_ta.read(i)
              x = x + 3
              out_ta_new = out_ta.write(i, x)
              return new_i, out_ta_new
  
          i_final, out_final = tf.while_loop(c, b, [i, output_ta])
          _ = tf.identity(i_final, name="i")
          _ = tf.identity(out_final.stack(), name="output_ta")
          input_names_with_port = ["input_1:0", "input_2:0"]
  
          output_names_with_port = ["i:0", "output_ta:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_map_fn(self):
          def fn0(elem):
              res = elem + elem * elem
              return res
  
          def fn1(elem):
              res = elem[0] * elem[1] + elem[0]
              return res
  
          x_val = 100 * np.random.random_sample([2, 10]).astype(np.float32)
          y_val = 100 * np.random.random_sample([2, 10]).astype(np.float32)
  
          # test fn0
          x = tf_placeholder(tf.float32, shape=x_val.shape, name="input_0")
          x_ = tf.identity(x)
          res_ = tf.map_fn(fn0, x_, dtype=tf.float32)
          _ = tf.identity(res_, name="output_0")
          input_names_with_port = ["input_0:0"]
          output_names_with_port = ["output_0:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
          tf_reset_default_graph()
  
          # test fn1
          x = tf_placeholder(tf.float32, shape=x_val.shape, name="input_0")
          y = tf_placeholder(tf.float32, shape=y_val.shape, name="input_1")
          x_ = tf.identity(x)
          y_ = tf.identity(y)
          res_ = tf.map_fn(fn1, (x_, y_), dtype=tf.float32)
          _ = tf.identity(res_, name="output_0")
          input_names_with_port = ["input_0:0", "input_1:0"]
          output_names_with_port = ["output_0:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_bidrectional_attention_wrapper_lstm_encoder(self):
          size = 30
          time_step = 3
          input_size = 4
          attn_size = size
          batch_size = 9
  
          # shape  [batch size, time step, size]
          # attention_state: usually the output of an RNN encoder.
          # This tensor should be shaped `[batch_size, max_time, ...]`
          encoder_time_step = time_step
          encoder_x_val = np.random.randn(encoder_time_step, input_size).astype('f')
          encoder_x_val = np.stack([encoder_x_val] * batch_size)
          encoder_x = tf_placeholder(tf.float32, encoder_x_val.shape, name="input_1")
          encoder_cell = tf.nn.rnn_cell.LSTMCell(size)
          attention_states, _ = tf.nn.dynamic_rnn(encoder_cell, encoder_x, dtype=tf.float32)
          # [9, 3, 30], [9, 30]
          attention_mechanism = tf.contrib.seq2seq.BahdanauAttention(attn_size,
                                                                     attention_states)
  
          match_input_fn = lambda curr_input, state: tf.concat([curr_input, state], axis=-1)
          cell = tf.nn.rnn_cell.LSTMCell(size)
          match_cell_fw = tf.contrib.seq2seq.AttentionWrapper(cell,
                                                              attention_mechanism,
                                                              attention_layer_size=attn_size,
                                                              cell_input_fn=match_input_fn,
                                                              output_attention=False)
          match_cell_bk = tf.contrib.seq2seq.AttentionWrapper(cell,
                                                              attention_mechanism,
                                                              attention_layer_size=attn_size,
                                                              cell_input_fn=match_input_fn,
                                                              output_attention=False)
  
          decoder_time_step = 6
          decoder_x_val = np.random.randn(decoder_time_step, batch_size, input_size).astype('f')
  
          decoder_x = tf_placeholder(tf.float32, decoder_x_val.shape, name="input_2")
          seq_length = tf_placeholder(tf.int32, (batch_size), name="input_3")
          (match_output_fw, match_output_bk), (match_state_fw, match_state_bk) = \
              tf.nn.bidirectional_dynamic_rnn(cell_fw=match_cell_fw,
                                              cell_bw=match_cell_bk,
                                              inputs=decoder_x,
                                              sequence_length=tf.identity(seq_length),
                                              dtype=tf.float32,
                                              time_major=True)
  
          matched_output = tf.concat([match_output_fw, match_output_bk], axis=-1)
          matched_state = tf.concat([match_state_fw.cell_state, match_state_bk.cell_state], -1)
  
          _ = tf.identity(matched_output, name="output_0")
          _ = tf.identity(matched_state, name="final_state")
  
          input_names_with_port = ["input_1:0", "input_2:0", "input_3:0"]
          output_names_with_port = ["output_0:0", "final_state:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_dynamic_decode_normal_stop(self):
          batch_size = 2
          num_units = 4
          vocab_size = 5
          embedding_size = 3
          go_token = 0
          end_token = 1
  
          embedding = tf.constant(np.ones([vocab_size, embedding_size], dtype=np.float32))
          state_val = np.reshape([np.ones([num_units], dtype=np.float32) * i for i in range(batch_size)],
                                 [batch_size, num_units])
          encoder_state = tf.nn.rnn_cell.LSTMStateTuple(state_val, state_val)
  
          cell_initializer = init_ops.constant_initializer(
              np.array([[-0.9592235, 0.42451382, 0.7437744, -0.54485345, -0.80763197,
                         0.19663906, -0.22738314, 0.7762785, 0.7464578, 0.27227187,
                         0.7661047, 0.3596425, -0.8528242, -0.89316916, -0.48946142,
                         0.87882376],
                        [0.86586094, -0.75018406, 0.25992537, -0.69368935, 0.2515502,
                         -0.26379275, 0.8954313, 0.5759742, -0.7753072, -0.4388857,
                         0.95751476, -0.82085776, -0.9467752, -0.37055635, -0.18570113,
                         -0.86504984],
                        [0.02305841, 0.3850248, 0.893692, -0.6866486, -0.83703446,
                         -0.9828961, 0.3989377, -0.59993076, 0.5330808, 0.6916566,
                         0.98468065, -0.6047034, 0.10823512, 0.34599304, -0.7834821,
                         -0.7852347],
                        [0.81643987, 0.31507468, -0.51369476, -0.12273741, 0.9701307,
                         -0.79669356, -0.34496522, -0.88750815, -0.17995334, 0.34707904,
                         -0.09201193, 0.5363934, -0.87229705, -0.5073328, -0.95894027,
                         0.5481839],
                        [-0.84093595, -0.2341497, -0.86047816, 0.43370056, -0.39073753,
                         0.37730122, 0.48026466, 0.3004985, -0.60727096, 0.9043884,
                         -0.37619448, 0.22490788, -0.03739262, 0.61672115, 0.478899,
                         -0.40780973],
                        [0.31202435, -0.22045255, -0.6087918, 0.95115066, 0.00199413,
                         -0.688287, -0.1103518, 0.4169519, 0.7913246, -0.9844644,
                         -0.6193857, 0.38659644, -0.4726901, -0.44781208, -0.5174744,
                         -0.605911],
                        [0.66771054, 0.34912825, 0.22297978, -0.4990945, 0.24057317,
                         -0.5540829, 0.92277217, 0.74939895, -0.35278273, -0.21587133,
                         -0.28613377, -0.8794241, -0.40119147, 0.67175174, -0.22741508,
                         0.37898326]], dtype=np.float32))
          dense_initializer = init_ops.constant_initializer(
              np.array([[0.56177187, -0.6233454, 0.73997784, 0.35032558, 0.6479795],
                        [0.6831174, -0.34233975, 0.39330363, 0.45177555, -0.49649096],
                        [-0.98890066, 0.6175642, 0.09800482, -0.6721206, 0.48805737],
                        [0.19671416, 0.2623148, 0.742548, 0.13555217, 0.56009054]], dtype=np.float32))
  
          cell = tf.nn.rnn_cell.LSTMCell(
              num_units=num_units,
              initializer=cell_initializer,
              state_is_tuple=True)
  
          helper = tf.contrib.seq2seq.GreedyEmbeddingHelper(
              embedding=embedding,
              start_tokens=tf.tile([go_token], [batch_size]),
              end_token=end_token)
  
          output_layer = tf.layers.Dense(vocab_size, kernel_initializer=dense_initializer)
          decoder = tf.contrib.seq2seq.BasicDecoder(
              cell=cell,
              helper=helper,
              initial_state=encoder_state,
              output_layer=output_layer)
  
          outputs, state, sequence_lengths = tf.contrib.seq2seq.dynamic_decode(
              decoder=decoder,
              maximum_iterations=6)
  
          _ = tf.identity(outputs.rnn_output, name="rnn_output")
          _ = tf.identity(outputs.sample_id, name="sample_id")
          _ = tf.identity(state, name="state")
          _ = tf.identity(sequence_lengths, name="sequence_lengths")
  
          output_names_with_port = [
              "rnn_output:0",
              # "sample_id:0",  # incomplete type support for Transpose on onnxruntime 0.2.1
              "state:0",
          ]
  
          self._run_test_case([], output_names_with_port)
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_while_loop_in_cond(self):
          x_val = np.array([1, 2, 3], dtype=np.float32)
          y_val = np.array([4, 5, 6], dtype=np.float32)
          x = tf_placeholder(tf.float32, x_val.shape, name="input_1")
          y = tf_placeholder(tf.float32, y_val.shape, name="input_2")
  
          def cond_graph():
              b = tf.constant(np.array([0], dtype=np.int32), dtype=tf.int32)
              # while_loop
              c = lambda y: tf.reduce_any(tf.less(y, 10))
              b = lambda i: tf.add(y, 1)
              return tf.while_loop(c, b, [y])
  
          res = tf.cond(x[0] < y[0], lambda: x, cond_graph, name="test_cond")
          _ = tf.identity(res, name="output")
  
          input_names_with_port = ["input_1:0", "input_2:0"]
          output_names_with_port = ["output:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
  
      @check_tf_max_version("1.15", "_run_test_case needs to supported tf-2")
      def test_cond_in_while_loop(self):
          i = tf.placeholder(tf.int32, (), name="input_1")
          inputs = tf.placeholder(tf.float32, (10,), name="input_2")
  
          inputs_2 = tf.identity(inputs)
          input_ta = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True).unstack(inputs_2)
          output_ta = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  
          c = lambda i, *_: tf.logical_and(tf.less(i, 10), i >= 0)
  
          def b(i, out_ta):
              new_i = tf.add(i, 1)
              x = input_ta.read(i)
              x = tf.cond(x > 0, lambda: x - 1, lambda: x + 3)
              out_ta_new = out_ta.write(i, x)
              return new_i, out_ta_new
  
          i_final, out_final = tf.while_loop(c, b, [i, output_ta])
          _ = tf.identity(i_final, name="i")
          _ = tf.identity(out_final.stack(), name="output_ta")
          input_names_with_port = ["input_1:0", "input_2:0"]
  
          output_names_with_port = ["i:0", "output_ta:0"]
          self._run_test_case(input_names_with_port, output_names_with_port)
  
  
  if __name__ == "__main__":
      unittest_main()
   # SPDX-License-Identifier: Apache-2.0
  
  
  """Unit Tests for gru."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from __future__ import unicode_literals
  
  import numpy as np
  import tensorflow as tf
  
  from tensorflow.python.ops import init_ops
  from tensorflow.python.ops import variable_scope
  from backend_test_base import Tf2OnnxBackendTestBase
  from common import unittest_main, check_gru_count, check_tf_max_version, check_opset_after_tf_version
  from tf2onnxnightly.tf_loader import is_tf2
  
  
  # pylint: disable=missing-docstring,invalid-name,unused-argument,using-constant-test,cell-var-from-loop
  
  if is_tf2():
      MultiRNNCell = tf.compat.v1.nn.rnn_cell.MultiRNNCell
      dynamic_rnn = tf.compat.v1.nn.dynamic_rnn
      bidirectional_dynamic_rnn = tf.compat.v1.nn.bidirectional_dynamic_rnn
  else:
      GRUBlockCell = tf.contrib.rnn.GRUBlockCell
      MultiRNNCell = tf.contrib.rnn.MultiRNNCell
      CudnnCompatibleGRUCell = tf.contrib.cudnn_rnn.CudnnCompatibleGRUCell
      dynamic_rnn = tf.nn.dynamic_rnn
      bidirectional_dynamic_rnn = tf.nn.bidirectional_dynamic_rnn
  
  
  # TODO: as a workaround, set batch_size to 1 for now to bypass a onnxruntime bug, revert it when the bug is fixed
  class CudnnCompatibleGRUTests(Tf2OnnxBackendTestBase):
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              # no scope
              cell = CudnnCompatibleGRUCell(units)
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          input_names_with_port = ["input_1:0"]
          feed_dict = {"input_1:0": x_val}
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-03, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_multiple_dynamic_gru(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              gru_output_list = []
              gru_cell_state_list = []
              # no scope
              cell = CudnnCompatibleGRUCell(units)
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32)
              gru_output_list.append(outputs)
              gru_cell_state_list.append(cell_state)
  
              # given scope
              cell = CudnnCompatibleGRUCell(units)
              with variable_scope.variable_scope("root1") as scope:
                  outputs, cell_state = dynamic_rnn(
                      cell,
                      x,
                      dtype=tf.float32,
                      sequence_length=[4],
                      scope=scope)
              gru_output_list.append(outputs)
              gru_cell_state_list.append(cell_state)
  
              return tf.identity(gru_output_list, name="output"), tf.identity(gru_cell_state_list, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06)
          # graph_validator=lambda g: check_gru_count(g, 2))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_seq_length_is_const(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.], [5., 5.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
  
              # no scope
              cell = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32,
                  sequence_length=[5])
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_seq_length_is_not_const(self):
          for np_dtype in [np.int32, np.int64, np.float32]:
              units = 5
              batch_size = 1
              x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.], [5., 5.]], dtype=np.float32)
              x_val = np.stack([x_val] * batch_size)
              y_val = np.array([5], dtype=np_dtype)
  
              def func(x, seq_length):
                  initializer = init_ops.constant_initializer(0.5)
                  # no scope
                  cell = CudnnCompatibleGRUCell(
                      units,
                      kernel_initializer=initializer)
                  outputs, cell_state = dynamic_rnn(
                      cell,
                      x,
                      dtype=tf.float32,
                      sequence_length=tf.identity(seq_length))
  
                  return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
              feed_dict = {"input_1:0": x_val, "input_2:0": y_val}
              input_names_with_port = ["input_1:0", "input_2:0"]
              output_names_with_port = ["output:0", "cell_state:0"]
              self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-03, atol=1e-06,
                                 graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_placeholder_input(self):
          units = 5
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.]], dtype=np.float32)
          x_val = np.stack([x_val] * 1)
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
              # no scope
              cell = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32)  # by default zero initializer is used
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-03, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_ch_zero_state_initializer(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.], [5., 5.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
              # no scope
              cell = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
  
              # defining initial state
              initial_state = cell.zero_state(batch_size, dtype=tf.float32)
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  initial_state=initial_state,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-03, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_random_weights(self):
          hidden_size = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.], [4., 4.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = tf.random_uniform_initializer(-1.0, 1.0)
  
              # no scope
              cell = CudnnCompatibleGRUCell(
                  hidden_size,
                  kernel_initializer=initializer)
  
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, 0.0001,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_single_dynamic_gru_random_weights2(self):
          hidden_size = 128
          batch_size = 1
          x_val = np.random.randn(1, 133).astype('f')
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = tf.random_uniform_initializer(0.0, 1.0)
              # no scope
              cell = CudnnCompatibleGRUCell(
                  hidden_size,
                  kernel_initializer=initializer)
  
              outputs, cell_state = dynamic_rnn(
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, 0.01,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_gru_output_consumed_only(self):
          units = 5
          batch_size = 6
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
          def func(x):
              initializer = tf.random_uniform_initializer(-1.0, 1.0)
              cell1 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
  
              outputs, _ = dynamic_rnn(
                  cell1,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, 0.0001,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_gru_state_consumed_only(self):
          units = 5
          batch_size = 6
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = tf.random_uniform_initializer(-1.0, 1.0)
              cell1 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
  
              _, cell_state = dynamic_rnn(
                  cell1,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=0.0001, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bigru(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
  
              # bigru, no scope
              cell1 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              cell2 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              outputs, cell_state = bidirectional_dynamic_rnn(
                  cell1,
                  cell2,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bigru_output_consumed_only(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
  
              # bigru, no scope
              cell1 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              cell2 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              outputs, _ = bidirectional_dynamic_rnn(
                  cell1,
                  cell2,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bigru_state_consumed_only(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
  
              # bigru, no scope
              cell1 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              cell2 = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              _, cell_state = bidirectional_dynamic_rnn(
                  cell1,
                  cell2,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bidirectional_but_one_gru(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
              initializer = init_ops.constant_initializer(0.5)
  
              # bigru, no scope
              cell = CudnnCompatibleGRUCell(
                  units,
                  kernel_initializer=initializer)
              outputs, cell_state = bidirectional_dynamic_rnn(
                  cell,
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output"), tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0", "cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bidirectional_but_one_gru_and_output_consumed_only(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
          def func(x):
              # bigru, no scope
              cell = CudnnCompatibleGRUCell(
                  units)
              outputs, _ = bidirectional_dynamic_rnn(
                  cell,
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(outputs, name="output")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["output:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
      @check_opset_after_tf_version("1.15", 10, "might need ReverseV2")
      @check_tf_max_version("1.15", "no CudnnCompatibleGRUCell in tf-2.x")
      def test_dynamic_bidirectional_but_one_gru_and_state_consumed_only(self):
          units = 5
          batch_size = 1
          x_val = np.array([[1., 1.], [2., 2.], [3., 3.]], dtype=np.float32)
          x_val = np.stack([x_val] * batch_size)
  
          def func(x):
  
              # bigru, no scope
              cell = CudnnCompatibleGRUCell(
                  units)
              _, cell_state = bidirectional_dynamic_rnn(
                  cell,
                  cell,
                  x,
                  dtype=tf.float32)
  
              return tf.identity(cell_state, name="cell_state")
  
          feed_dict = {"input_1:0": x_val}
          input_names_with_port = ["input_1:0"]
          output_names_with_port = ["cell_state:0"]
          self.run_test_case(func, feed_dict, input_names_with_port, output_names_with_port, rtol=1e-3, atol=1e-06,
                             graph_validator=lambda g: check_gru_count(g, 1))
  
  
  if __name__ == '__main__':
      unittest_main()
  
apis: numpy.ones tensorflow.nn.dynamic_rnn tensorflow.concat numpy.stack tensorflow.nn.rnn_cell.LSTMStateTuple tensorflow.identity tensorflow.contrib.seq2seq.dynamic_decode tensorflow.compat.v1.disable_eager_execution tensorflow.nn.rnn_cell.LSTMCell tensorflow.less tensorflow.import_graph_def tensorflow.contrib.seq2seq.AttentionWrapper tensorflow.cond tensorflow.layers.Dense tensorflow.contrib.seq2seq.BahdanauAttention tensorflow.contrib.seq2seq.BasicDecoder tensorflow.python.ops.variables.global_variables_initializer tensorflow.while_loop tensorflow.tile tensorflow.graph_util.convert_variables_to_constants numpy.random.random_sample tensorflow.placeholder tensorflow.map_fn numpy.random.randn tensorflow.add tensorflow.TensorArray numpy.array
apis: numpy.stack tensorflow.identity numpy.random.randn tensorflow.python.ops.variable_scope.variable_scope tensorflow.random_uniform_initializer numpy.array tensorflow.python.ops.init_ops.constant_initializer