docstr: Tests that the tensorboard writer writes the correct scalars to SummaryWriter
  iff is_primary() is True.
docstr_tok: ["Tests","that","the","tensorboard","writer","writes","the","correct","scalars","to","SummaryWriter","iff","is_primary","(",")","is","True","."]
code: def test_writer(self, mock_is_primary_func: mock.MagicMock) -> None:
      """
      Tests that the tensorboard writer writes the correct scalars to SummaryWriter
      iff is_primary() is True.
      """
      for phase_idx, primary in product([0, 1, 2], [True, False]):
          train, phase_type = (
              (True, "train") if phase_idx % 2 == 0 else (False, "test")
          )
          mock_is_primary_func.return_value = primary
  
          # set up the task and state
          config = get_test_task_config()
          config["dataset"]["train"]["batchsize_per_replica"] = 2
          config["dataset"]["test"]["batchsize_per_replica"] = 5
          task = build_task(config)
          task.prepare()
          task.advance_phase()
          task.phase_idx = phase_idx
          task.train = train
  
          losses = [1.23, 4.45, 12.3, 3.4]
          sample_fetch_times = [1.1, 2.2, 3.3, 2.2]
  
          summary_writer = SummaryWriter(self.base_dir)
          # create a spy on top of summary_writer
          summary_writer = mock.MagicMock(wraps=summary_writer)
  
          # create a loss lr tensorboard hook
          tensorboard_plot_hook = TensorboardPlotHook(summary_writer)
  
          # run the hook in the correct order
          tensorboard_plot_hook.on_phase_start(task)
  
          # test tasks which do not pass the sample_fetch_times as well
          disable_sample_fetch_times = phase_idx == 0
  
          for loss, sample_fetch_time in zip(
              losses, sample_fetch_times
          ):
              task.losses.append(loss)
              step_data = (
                  {}
                  if disable_sample_fetch_times
                  else {"sample_fetch_time": sample_fetch_time}
              )
              task.last_batch = LastBatchInfo(
                  None, None, None, None, step_data
              )
              tensorboard_plot_hook.on_step(task)
  
          tensorboard_plot_hook.on_phase_end(task)
  
          if primary:
              # add_scalar() should have been called with the right scalars
              if train:
                  learning_rate_key = f"Learning Rate/{phase_type}"
                  summary_writer.add_scalar.assert_any_call(
                      learning_rate_key,
                      mock.ANY,
                      global_step=mock.ANY,
                      walltime=mock.ANY,
                  )
              avg_loss_key = f"Losses/{phase_type}"
              summary_writer.add_scalar.assert_any_call(
                  avg_loss_key, mock.ANY, global_step=mock.ANY
              )
              for meter in task.meters:
                  for name in meter.value:
                      meter_key = (
                          f"Meters/{phase_type}/{meter.name}/{name}"
                      )
                      summary_writer.add_scalar.assert_any_call(
                          meter_key, mock.ANY, global_step=mock.ANY
                      )
              if step_data:
                  summary_writer.add_scalar.assert_any_call(
                      f"Speed/{phase_type}/cumulative_sample_fetch_time",
                      mock.ANY,
                      global_step=mock.ANY,
                      walltime=mock.ANY,
                  )
          else:
              # add_scalar() shouldn't be called since is_primary() is False
              summary_writer.add_scalar.assert_not_called()
          summary_writer.add_scalar.reset_mock()
  
file_path: ClassyVision/test/manual/hooks_tensorboard_plot_hook_test.py
imports: import copy
  import shutil
  import tempfile
  import unittest.mock
  from itertools import product
  from classy_vision.hooks import TensorboardPlotHook
  from classy_vision.optim.param_scheduler import ClassyParamScheduler, UpdateInterval
  from classy_vision.tasks import build_task
  from classy_vision.tasks.classification_task import LastBatchInfo
  from classy_vision.trainer import LocalTrainer
  from test.generic.config_utils import get_test_mlp_task_config, get_test_task_config
  from test.generic.hook_test_utils import HookTestBase
  from torch.utils.tensorboard import SummaryWriter
label_desc: Use image node_id 4 for calling the TestTensorboardPlotHook obj's underlying member method code with example usage: obj.test_writer(mock_is_primary_func) without return types