code: def test_evaluate_q_hvkg(self):
      tkwargs = {"device": self.device}
      num_pareto = 3
      for dtype, acqf_class in product(
          (torch.float, torch.double),
          (
              qHypervolumeKnowledgeGradient,
              qMultiFidelityHypervolumeKnowledgeGradient,
          ),
      ):
          tkwargs["dtype"] = dtype
          # basic test
          n_f = 4
          mean = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
          variance = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
          mfm = MockModel(MockPosterior(mean=mean, variance=variance))
          ref_point = torch.zeros(2, **tkwargs)
          models = [
              SingleTaskGP(
                  torch.rand(2, 1, **tkwargs),
                  torch.rand(2, 1, **tkwargs),
              ),
              SingleTaskGP(
                  torch.rand(4, 1, **tkwargs),
                  torch.rand(4, 1, **tkwargs),
              ),
          ]
          model = ModelListGP(*models)
          if acqf_class == qMultiFidelityHypervolumeKnowledgeGradient:
              mf_kwargs = {"target_fidelities": {-1: 1.0}}
          else:
              mf_kwargs = {}
  
          with mock.patch.object(
              ModelListGP, "fantasize", return_value=mfm
          ) as patch_f:
              with mock.patch(
                  NO, new_callable=mock.PropertyMock
              ) as mock_num_outputs:
                  mock_num_outputs.return_value = 2
  
                  qHVKG = acqf_class(
                      model=model,
                      num_fantasies=n_f,
                      ref_point=ref_point,
                      num_pareto=num_pareto,
                      **mf_kwargs,
                  )
                  X = torch.rand(n_f * num_pareto + 1, 1, **tkwargs)
                  val = qHVKG(X)
                  patch_f.assert_called_once()
                  cargs, ckwargs = patch_f.call_args
                  self.assertEqual(
                      ckwargs["X"].shape, torch.Size([1, 1, 1])
                  )
          expected_hv = (
              DominatedPartitioning(
                  Y=mean.squeeze(1), ref_point=ref_point
              )
              .compute_hypervolume()
              .mean()
          )
          self.assertAllClose(val.item(), expected_hv.item(), atol=1e-4)
          self.assertTrue(
              torch.equal(
                  qHVKG.extract_candidates(X),
                  X[..., : -n_f * num_pareto, :],
              )
          )
  
          # batched evaluation
          b = 2
          mean = torch.rand(n_f, b, num_pareto, 2, **tkwargs)
          variance = torch.rand(n_f, b, num_pareto, 2, **tkwargs)
          mfm = MockModel(MockPosterior(mean=mean, variance=variance))
          X = torch.rand(b, n_f * num_pareto + 1, 1, **tkwargs)
          with mock.patch.object(
              ModelListGP, "fantasize", return_value=mfm
          ) as patch_f:
              with mock.patch(
                  NO, new_callable=mock.PropertyMock
              ) as mock_num_outputs:
                  mock_num_outputs.return_value = 2
                  qHVKG = acqf_class(
                      model=model,
                      num_fantasies=n_f,
                      ref_point=ref_point,
                      num_pareto=num_pareto,
                      **mf_kwargs,
                  )
                  val = qHVKG(X)
                  patch_f.assert_called_once()
                  cargs, ckwargs = patch_f.call_args
                  self.assertEqual(
                      ckwargs["X"].shape, torch.Size([b, 1, 1])
                  )
          expected_hv = (
              DominatedPartitioning(
                  Y=mean.view(-1, num_pareto, 2), ref_point=ref_point
              )
              .compute_hypervolume()
              .view(n_f, b)
              .mean(dim=0)
          )
          self.assertAllClose(val, expected_hv, atol=1e-4)
          self.assertTrue(
              torch.equal(
                  qHVKG.extract_candidates(X),
                  X[..., : -n_f * num_pareto, :],
              )
          )
          # pending points and current value
          X_pending = torch.rand(2, 1, **tkwargs)
          X_pending_evaluation_mask = torch.eye(
              2, device=self.device
          ).bool()
          X_evaluation_mask = torch.tensor(
              [[False, True]], dtype=torch.bool, device=self.device
          )
          mean = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
          variance = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
          mfm = MockModel(MockPosterior(mean=mean, variance=variance))
          current_value = torch.tensor(0.0, **tkwargs)
          X = torch.rand(n_f * num_pareto + 1, 1, **tkwargs)
          cost_aware_utility = InverseCostWeightedUtility(
              cost_model=GenericDeterministicModel(
                  cost_fn, num_outputs=2
              )
          )
          with mock.patch.object(
              ModelListGP, "fantasize", return_value=mfm
          ) as patch_f:
              with mock.patch(
                  NO, new_callable=mock.PropertyMock
              ) as mock_num_outputs:
                  mock_num_outputs.return_value = 2
                  qHVKG = acqf_class(
                      model=model,
                      num_fantasies=n_f,
                      X_pending=X_pending,
                      X_pending_evaluation_mask=X_pending_evaluation_mask,
                      X_evaluation_mask=X_evaluation_mask,
                      current_value=current_value,
                      ref_point=ref_point,
                      num_pareto=num_pareto,
                      cost_aware_utility=cost_aware_utility,
                      **mf_kwargs,
                  )
                  val = qHVKG(X)
                  patch_f.assert_called_once()
                  expected_eval_mask = torch.cat(
                      [X_evaluation_mask, X_pending_evaluation_mask],
                      dim=0,
                  )
                  cargs, ckwargs = patch_f.call_args
                  self.assertEqual(
                      ckwargs["X"].shape, torch.Size([1, 3, 1])
                  )
                  self.assertTrue(
                      torch.equal(
                          ckwargs["evaluation_mask"], expected_eval_mask
                      )
                  )
          expected_hv = (
              DominatedPartitioning(
                  Y=mean.squeeze(1), ref_point=ref_point
              )
              .compute_hypervolume()
              .mean(dim=0)
          )
          # divide by 3 because the cost is 3 per design
          expected = (expected_hv.mean() - current_value).item() / 3
          self.assertAllClose(val.item(), expected, atol=1e-4)
          self.assertTrue(
              torch.equal(
                  qHVKG.extract_candidates(X),
                  X[..., : -n_f * num_pareto, :],
              )
          )
          # test that cost-weighted HVKG is clamped
          with mock.patch.object(
              ModelListGP, "fantasize", return_value=mfm
          ) as patch_f:
              with mock.patch(
                  NO, new_callable=mock.PropertyMock
              ) as mock_num_outputs:
                  mock_num_outputs.return_value = 2
                  qHVKG = acqf_class(
                      model=model,
                      num_fantasies=n_f,
                      X_pending=X_pending,
                      X_pending_evaluation_mask=X_pending_evaluation_mask,
                      X_evaluation_mask=X_evaluation_mask,
                      current_value=torch.tensor(1000, **tkwargs),
                      ref_point=ref_point,
                      num_pareto=num_pareto,
                      cost_aware_utility=cost_aware_utility,
                      **mf_kwargs,
                  )
                  val = qHVKG(X)
                  self.assertEqual(val.item(), 0.0)
  
          # test mfkg
          if acqf_class == qMultiFidelityHypervolumeKnowledgeGradient:
              mean = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
              variance = torch.rand(n_f, 1, num_pareto, 2, **tkwargs)
              mfm = MockModel(
                  MockPosterior(mean=mean, variance=variance)
              )
              current_value = torch.rand(1, **tkwargs)
              X = torch.rand(n_f * num_pareto + 1, 1, **tkwargs)
              with mock.patch(
                  "botorch.acquisition.multi_objective."
                  "hypervolume_knowledge_gradient._get_hv_value_function",
                  wraps=_get_hv_value_function,
              ) as mock_get_value_func:
                  with mock.patch.object(
                      ModelListGP, "fantasize", return_value=mfm
                  ) as patch_f:
                      with mock.patch(
                          NO, new_callable=mock.PropertyMock
                      ) as mock_num_outputs:
                          mock_num_outputs.return_value = 2
                          qHVKG = acqf_class(
                              model=model,
                              num_fantasies=n_f,
                              current_value=current_value,
                              ref_point=ref_point,
                              num_pareto=num_pareto,
                              **mf_kwargs,
                          )
                          val = qHVKG(X)
                          self.assertIsNotNone(
                              mock_get_value_func.call_args_list[0][1][
                                  "project"
                              ]
                          )
  
          # test objective (inner MC sampling)
          mean = torch.rand(n_f, 1, num_pareto, 3, **tkwargs)
          samples = mean + 1
          variance = torch.rand(n_f, 1, num_pareto, 3, **tkwargs)
          mfm = MockModel(
              MockPosterior(
                  mean=mean, variance=variance, samples=samples
              )
          )
          models = [
              SingleTaskGP(
                  torch.rand(2, 1, **tkwargs),
                  torch.rand(2, 1, **tkwargs),
              ),
              SingleTaskGP(
                  torch.rand(4, 1, **tkwargs),
                  torch.rand(4, 1, **tkwargs),
              ),
              SingleTaskGP(
                  torch.rand(5, 1, **tkwargs),
                  torch.rand(5, 1, **tkwargs),
              ),
          ]
          model = ModelListGP(*models)
          for num_objectives in (2, 3):
              # test using 1) a botorch objective that only uses 2 out of
              # 3 outcomes as objectives, 2) a botorch objective that uses
              # all 3 outcomes as objectives
              objective = (
                  IdentityMCMultiOutputObjective(outcomes=[0, 1])
                  if num_objectives == 2
                  else GenericMCMultiOutputObjective(lambda Y, X: 2 * Y)
              )
  
              ref_point = torch.zeros(num_objectives, **tkwargs)
              X = torch.rand(n_f * num_pareto + 1, 1, **tkwargs)
  
              for use_posterior_mean in (True, False):
                  with mock.patch.object(
                      ModelListGP, "fantasize", return_value=mfm
                  ) as patch_f:
                      with mock.patch(
                          NO, new_callable=mock.PropertyMock
                      ) as mock_num_outputs:
                          mock_num_outputs.return_value = 3
                          qHVKG = acqf_class(
                              model=model,
                              num_fantasies=n_f,
                              objective=objective,
                              ref_point=ref_point,
                              num_pareto=num_pareto,
                              use_posterior_mean=use_posterior_mean,
                              **mf_kwargs,
                          )
                          val = qHVKG(X)
                          patch_f.assert_called_once()
                          cargs, ckwargs = patch_f.call_args
                          self.assertEqual(
                              ckwargs["X"].shape, torch.Size([1, 1, 1])
                          )
                  Ys = mean if use_posterior_mean else samples
                  objs = objective(Ys.squeeze(1)).view(
                      -1, num_pareto, num_objectives
                  )
                  if num_objectives == 2:
                      expected_hv = (
                          DominatedPartitioning(
                              Y=objs, ref_point=ref_point
                          )
                          .compute_hypervolume()
                          .mean()
                          .item()
                      )
                  else:
                      # batch box decomposition don't support > 2 objectives
                      objs = objective(Ys).view(
                          -1, num_pareto, num_objectives
                      )
                      expected_hv = np.mean(
                          [
                              DominatedPartitioning(
                                  Y=obj, ref_point=ref_point
                              )
                              .compute_hypervolume()
                              .mean()
                              .item()
                              for obj in objs
                          ]
                      )
                  self.assertAllClose(
                      val.item(), expected_hv, atol=1e-4
                  )
                  self.assertTrue(
                      torch.equal(
                          qHVKG.extract_candidates(X),
                          X[..., : -n_f * num_pareto, :],
                      )
                  )
  
imports: from itertools import product
  from unittest import mock
  import numpy
  import torch
  from botorch.acquisition.cost_aware import InverseCostWeightedUtility
  from botorch.acquisition.multi_objective.hypervolume_knowledge_gradient import _get_hv_value_function, _split_hvkg_fantasy_points, qHypervolumeKnowledgeGradient, qMultiFidelityHypervolumeKnowledgeGradient
  from botorch.acquisition.multi_objective.objective import GenericMCMultiOutputObjective, IdentityMCMultiOutputObjective
  from botorch.exceptions.errors import UnsupportedError
  from botorch.models.deterministic import GenericDeterministicModel
  from botorch.models.gp_regression import SingleTaskGP
  from botorch.models.model_list_gp_regression import ModelListGP
  from botorch.sampling.list_sampler import ListSampler
  from botorch.sampling.normal import SobolQMCNormalSampler
  from botorch.utils.multi_objective.box_decompositions.dominated import DominatedPartitioning
  from botorch.utils.testing import BotorchTestCase, MockModel, MockPosterior
label_desc: Use image node_id 2 for calling the TestHypervolumeKnowledgeGradient obj's underlying member method code with example usage: obj.test_evaluate_q_hvkg() without return types