class_docstr: Compilation and learning tests for the Fitted-Q Evaluation model
class_docstr_tok: ["Compilation","and","learning","tests","for","the","Fitted-Q","Evaluation","model"]
docstr: Compilation tests for FQETorchModel.
  
  (1) Check that it does not modify the underlying batch during training
  (2) Check that the stopping criteria from FQE are working correctly
  (3) Check that using fqe._compute_action_probs equals brute force
  iterating over all actions with policy.compute_log_likelihoods
docstr_tok: ["Compilation","tests","for","FQETorchModel",".","(","1",")","Check","that","it","does","not","modify","the","underlying","batch","during","training","(","2",")","Check","that","the","stopping","criteria","from","FQE","are","working","correctly","(","3",")","Check","that","using","fqe._compute_action_probs","equals","brute","force","iterating","over","all","actions","with","policy.compute_log_likelihoods"]
code: def test_fqe_compilation_and_stopping(self):
      """Compilation tests for FQETorchModel.
  
      (1) Check that it does not modify the underlying batch during training
      (2) Check that the stopping criteria from FQE are working correctly
      (3) Check that using fqe._compute_action_probs equals brute force
      iterating over all actions with policy.compute_log_likelihoods
      """
      fqe = FQETorchModel(
          policy=self.policy,
          gamma=self.gamma,
      )
      tmp_batch = copy.deepcopy(self.batch)
      losses = fqe.train(self.batch)
  
      # Make sure FQETorchModel.train() does not modify the batch
      check(tmp_batch, self.batch)
  
      # Make sure FQE stopping criteria are respected
      assert (
          len(losses) == fqe.n_iters
          or losses[-1] < fqe.min_loss_threshold
      ), (
          f"FQE.train() terminated early in {len(losses)} steps with final loss"
          f"{losses[-1]} for n_iters: {fqe.n_iters} and "
          f"min_loss_threshold: {fqe.min_loss_threshold}"
      )
  
      # Test fqe._compute_action_probs against "brute force" method
      # of computing log_prob for each possible action individually
      # using policy.compute_log_likelihoods
      obs = torch.tensor(self.batch["obs"], device=fqe.device)
      action_probs = fqe._compute_action_probs(obs)
      action_probs = convert_to_numpy(action_probs)
  
      tmp_probs = []
      for act in range(fqe.policy.action_space.n):
          tmp_actions = np.zeros_like(self.batch["actions"]) + act
          log_probs = self.policy.compute_log_likelihoods(
              actions=tmp_actions,
              obs_batch=self.batch["obs"],
          )
          tmp_probs.append(np.exp(log_probs))
      tmp_probs = np.stack(tmp_probs).T
      check(action_probs, tmp_probs, decimals=3)
  
imports: from typing import TYPE_CHECKING, Tuple
  import copy
  import gymnasium
  import numpy
  import os
  import shutil
  import tempfile
  import pandas
  from pathlib import Path
  import unittest
  import ray
  from ray.data import read_json
  from ray.rllib.algorithms.algorithm import Algorithm
  from ray.rllib.algorithms.dqn import DQNConfig
  from ray.rllib.examples.env.cliff_walking_wall_env import CliffWalkingWallEnv
  from ray.rllib.examples.policy.cliff_walking_wall_policy import CliffWalkingWallPolicy
  from ray.rllib.offline.dataset_reader import DatasetReader
  from ray.rllib.offline.estimators import DirectMethod, DoublyRobust, ImportanceSampling, WeightedImportanceSampling
  from ray.rllib.offline.estimators.fqe_torch_model import FQETorchModel
  from ray.rllib.policy.sample_batch import SampleBatch, concat_samples
  from ray.rllib.utils.framework import try_import_torch
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.test_utils import check
label_desc: Use image node_id 3 for calling the TestFQE obj's underlying member method code with example usage: obj.test_fqe_compilation_and_stopping() without return types