docstr: Tests V-trace against ground truth data calculated in python.
docstr_tok: ["Tests","V-trace","against","ground","truth","data","calculated","in","python","."]
code: def test_from_importance_weights(self, name, batch_size):
      """Tests V-trace against ground truth data calculated in python."""
      seq_len = 5
  
      # Create log_rhos such that rho will span from near-zero to above the
      # clipping thresholds. In particular, calculate log_rhos in [-2.5, 2.5),
      # so that rho is in approx [0.08, 12.2).
      log_rhos = _shaped_arange(seq_len, batch_size) / (
          batch_size * seq_len
      )
      log_rhos = 5 * (log_rhos - 0.5)  # [0.0, 1.0) -> [-2.5, 2.5).
      # Fake behaviour_actions_log_probs, target_actions_log_probs
      target_actions_log_probs = log_rhos + 1.0
      behaviour_actions_log_probs = np.ones(
          shape=log_rhos.shape, dtype="float32"
      )
  
      values = {
          "behaviour_actions_log_probs": behaviour_actions_log_probs,
          "target_actions_log_probs": target_actions_log_probs,
          # T, B where B_i: [0.9 / (i+1)] * T
          "discounts": np.array(
              [
                  [0.9 / (b + 1) for b in range(batch_size)]
                  for _ in range(seq_len)
              ],
              dtype=np.float32,
          ),
          "rewards": _shaped_arange(seq_len, batch_size),
          "values": _shaped_arange(seq_len, batch_size) / batch_size,
          "bootstrap_value": _shaped_arange(batch_size) + 1.0,
          "clip_rho_threshold": 3.7,
          "clip_pg_rho_threshold": 2.2,
      }
  
      # Calculated by numpy/python
      ground_truth_v = _ground_truth_calculation(**values)
  
      # Calculated by Fluid
      test_program = fluid.Program()
      with fluid.program_guard(test_program):
          behaviour_actions_log_probs_input = layers.data(
              name="behaviour_actions_log_probs",
              shape=[seq_len, batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          target_actions_log_probs_input = layers.data(
              name="target_actions_log_probs",
              shape=[seq_len, batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          discounts_input = layers.data(
              name="discounts",
              shape=[seq_len, batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          rewards_input = layers.data(
              name="rewards",
              shape=[seq_len, batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          values_input = layers.data(
              name="values",
              shape=[seq_len, batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          bootstrap_value_input = layers.data(
              name="bootstrap_value",
              shape=[batch_size],
              dtype="float32",
              append_batch_size=False,
          )
          fluid_inputs = {
              "behaviour_actions_log_probs": behaviour_actions_log_probs_input,
              "target_actions_log_probs": target_actions_log_probs_input,
              "discounts": discounts_input,
              "rewards": rewards_input,
              "values": values_input,
              "bootstrap_value": bootstrap_value_input,
              "clip_rho_threshold": 3.7,
              "clip_pg_rho_threshold": 2.2,
          }
          output = vtrace.from_importance_weights(**fluid_inputs)
  
      self.executor.run(fluid.default_startup_program())
      feed = copy.copy(values)
      del feed["clip_rho_threshold"]
      del feed["clip_pg_rho_threshold"]
      [output_vs, output_pg_advantage] = self.executor.run(
          test_program,
          feed=feed,
          fetch_list=[output.vs, output.pg_advantages],
      )
  
      np.testing.assert_almost_equal(ground_truth_v.vs, output_vs, 5)
      np.testing.assert_almost_equal(
          ground_truth_v.pg_advantages, output_pg_advantage, 5
      )
  
file_path: PARL/parl/algorithms/fluid/impala/tests/vtrace_test_fluid.py
imports: import copy
  import numpy
  import unittest
  from parl import layers
  from paddle import fluid
  from parameterized import parameterized
  from parl.algorithms.fluid.impala import vtrace
  from parl.utils import get_gpu_count
label_desc: Use image node_id 2 for calling the VtraceTest obj's underlying member method code with example usage: obj.test_from_importance_weights(name, batch_size) without return types