code: def _ground_truth_vtrace_calculation(
      discounts: np.ndarray,
      log_rhos: np.ndarray,
      rewards: np.ndarray,
      values: np.ndarray,
      bootstrap_value: np.ndarray,
      clip_rho_threshold: float,
      clip_pg_rho_threshold: float,
  ):
      """Calculates the ground truth for V-trace in Python/Numpy.
  
      NOTE:
      The discount, log_rhos, rewards, values, and bootstrap_value are all assumed to
      come from trajectories of experience. Typically batches of trajectories could be
      thought of as having the shape [B, T] where B is the batch dimension, and T is
      the timestep dimension. Computing vtrace returns requires that the data is time
      major, meaning that it has the shape [T, B]. One can use a function like
      `make_time_major` to properly format their discount, log_rhos, rewards, values,
      and bootstrap_value before calling _ground_truth_vtrace_calculation.
  
      Args:
          discounts: Array of shape [T*B] of discounts. T is the lenght of the trajectory
              or sequence and B is the batch size.
              NOTE: The discount will be equal to gamma, the discount factor when the
              timestep that the discount is being applied to is not a terminal timestep.
          log_rhos: Array of shape [T*B] of log likelihood ratios of target action
              probabilities to behavior action probabilities.
          rewards: Array of shape [T*B] of rewards.
          values: Array of shape [T*B] of the value function estimated for every timestep
              in a batch.
          bootstrap_value: Array of shape [T] of the value function estimated at the last
              timestep for each trajectory in the batch.
          clip_rho_threshold: The threshold for clipping the importance weights.
          clip_pg_rho_threshold: The threshold for clipping the importance weights for
              the policy gradient loss.
  
      Returns:
          The v-trace adjusted values and the policy gradient advantages.
  
      """
      vs = []
      seq_len = len(discounts)
      rhos = np.exp(log_rhos)
      cs = np.minimum(rhos, 1.0)
      clipped_rhos = rhos
      if clip_rho_threshold:
          clipped_rhos = np.minimum(rhos, clip_rho_threshold)
      clipped_pg_rhos = rhos
      if clip_pg_rho_threshold:
          clipped_pg_rhos = np.minimum(rhos, clip_pg_rho_threshold)
  
      # This is a very inefficient way to calculate the V-trace ground truth.
      # We calculate it this way because it is close to the mathematical notation
      # of
      # V-trace.
      # v_s = V(x_s)
      #       + \sum^{T-1}_{t=s} \gamma^{t-s}
      #         * \prod_{i=s}^{t-1} c_i
      #         * \rho_t (r_t + \gamma V(x_{t+1}) - V(x_t))
      # Note that when we take the product over c_i, we write `s:t` as the
      # notation
      # of the paper is inclusive of the `t-1`, but Python is exclusive.
      # Also note that np.prod([]) == 1.
      values_t_plus_1 = np.concatenate(
          [values[1:], bootstrap_value[None, :]], axis=0
      )
      for s in range(seq_len):
          v_s = np.copy(values[s])  # Very important copy.
          for t in range(s, seq_len):
              v_s += (
                  np.prod(discounts[s:t], axis=0)
                  * np.prod(cs[s:t], axis=0)
                  * clipped_rhos[t]
                  * (
                      rewards[t]
                      + discounts[t] * values_t_plus_1[t]
                      - values[t]
                  )
              )
          vs.append(v_s)
      vs = np.stack(vs, axis=0)
      pg_advantages = clipped_pg_rhos * (
          rewards
          + discounts
          * np.concatenate([vs[1:], bootstrap_value[None, :]], axis=0)
          - values
      )
      return vs, pg_advantages
  
file_path: ray/rllib/algorithms/impala/tests/test_vtrace.py
imports: from gymnasium.spaces import Box
  import numpy
  import unittest
  from ray.rllib.algorithms.impala import vtrace_tf
  from ray.rllib.algorithms.impala import vtrace_torch
  from ray.rllib.utils.framework import try_import_tf, try_import_torch
  from ray.rllib.utils.numpy import softmax
  from ray.rllib.utils.test_utils import check, framework_iterator
label_desc: Use image node_id 1 for calling a global function with example usage: _ground_truth_vtrace_calculation(discounts, log_rhos, rewards, values, bootstrap_value, clip_rho_threshold, clip_pg_rho_threshold) and returns: vs,  pg_advantages