docstr: Ensure that observations and rewards change in consecutive calls.
docstr_tok: ["Ensure","that","observations","and","rewards","change","in","consecutive","calls","."]
code: def testObservationAndRewardsVary(
      self, observation_shape, action_shape, batch_size, seed
  ):
      """Ensure that observations and rewards change in consecutive calls."""
      interval = 4
  
      env = get_deterministic_gaussian_non_stationary_environment(
          observation_shape, action_shape, batch_size, interval
      )
      tf.compat.v1.set_random_seed(seed)
      self.evaluate(tf.compat.v1.global_variables_initializer())
  
      observation_to_reward_samples = []
      additive_reward_samples = []
      env_time = env._env_time
      observation_to_reward = (
          env._environment_dynamics._current_observation_to_reward
      )
      additive_reward = (
          env._environment_dynamics._current_additive_reward
      )
  
      if tf.executing_eagerly():
          for t in range(0, 10):
              ts = env.reset()
              reward = env.step(tf.zeros([batch_size])).reward
  
              (
                  observation_to_reward_sample,
                  additive_reward_sample,
                  env_time_sample,
              ) = self.evaluate(
                  [observation_to_reward, additive_reward, env_time]
              )
              observation_to_reward_samples.append(
                  observation_to_reward_sample
              )
              additive_reward_samples.append(additive_reward_sample)
              self.assertEqual(env_time_sample, (t + 1) * batch_size)
  
      else:
          ts = env.reset()
          reward = env.step(tf.zeros([batch_size])).reward
  
          for t in range(0, 10):
              unused_ts_sample = self.evaluate(ts)
              unused_reward_sample = self.evaluate(reward)
  
              (
                  observation_to_reward_sample,
                  additive_reward_sample,
                  env_time_sample,
              ) = self.evaluate(
                  [observation_to_reward, additive_reward, env_time]
              )
  
              observation_to_reward_samples.append(
                  observation_to_reward_sample
              )
              additive_reward_samples.append(additive_reward_sample)
              self.assertEqual(env_time_sample, (t + 1) * batch_size)
  
      for t in range(0, 10):
          self.assertAllClose(
              observation_to_reward_samples[
                  int(t - t % (interval / batch_size))
              ],
              observation_to_reward_samples[t],
          )
          self.assertAllClose(
              additive_reward_samples[
                  int(t - t % (interval / batch_size))
              ],
              additive_reward_samples[t],
          )
  
      for t in range(
          interval // batch_size, 10, interval // batch_size
      ):
          self.assertNotAllClose(
              observation_to_reward_samples[int(t) - 1],
              observation_to_reward_samples[int(t)],
          )
          self.assertNotAllClose(
              additive_reward_samples[int(t) - 1],
              additive_reward_samples[int(t)],
          )
  
file_path: agents/tf_agents/bandits/environments/piecewise_stochastic_environment_test.py
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from absl.testing import parameterized
  import tensorflow
  import tensorflow_probability
  from tf_agents.bandits.environments import piecewise_stochastic_environment
label_desc: Use image node_id 1 for calling the PiecewiseStochasticEnvironmentTest obj's underlying member method code with example usage: obj.testObservationAndRewardsVary(observation_shape, action_shape, batch_size, seed) without return types