code: def test_log_works_in_train_callback(tmpdir):
      """Tests that log can be called within callback."""
  
      class TestCallback(callbacks.Callback):
          count = 0
          choices = [False, True]
  
          # used to compute expected values
          logged_values = collections.defaultdict(list)
          call_counter = collections.Counter()
          logged_arguments = {}
  
          def make_logging(
              self, pl_module, func_name, on_steps, on_epochs, prob_bars
          ):
              self.call_counter.update([func_name])
  
              for idx, (on_step, on_epoch, prog_bar) in enumerate(
                  itertools.product(on_steps, on_epochs, prob_bars)
              ):
                  fx = f"{func_name}_{idx}"
                  if not on_step and not on_epoch:
                      with pytest.raises(
                          MisconfigurationException,
                          match="is not useful",
                      ):
                          pl_module.log(
                              fx,
                              self.count,
                              on_step=on_step,
                              on_epoch=on_epoch,
                          )
                      continue
                  pl_module.log(
                      fx,
                      self.count,
                      on_step=on_step,
                      on_epoch=on_epoch,
                      prog_bar=prog_bar,
                  )
                  self.logged_values[fx].append(self.count)
                  self.logged_arguments[fx] = {
                      "on_step": on_step,
                      "on_epoch": on_epoch,
                      "prog_bar": prog_bar,
                  }
                  self.count += 1
  
          def on_train_start(self, _, pl_module):
              self.make_logging(
                  pl_module,
                  "on_train_start",
                  on_steps=[False],
                  on_epochs=[True],
                  prob_bars=self.choices,
              )
  
          def on_train_epoch_start(self, _, pl_module):
              self.make_logging(
                  pl_module,
                  "on_train_epoch_start",
                  on_steps=[False],
                  on_epochs=[True],
                  prob_bars=self.choices,
              )
  
          def on_train_batch_start(self, _, pl_module, *__):
              self.make_logging(
                  pl_module,
                  "on_train_batch_start",
                  on_steps=self.choices,
                  on_epochs=self.choices,
                  prob_bars=self.choices,
              )
  
          def on_train_batch_end(self, _, pl_module, *__):
              self.make_logging(
                  pl_module,
                  "on_train_batch_end",
                  on_steps=self.choices,
                  on_epochs=self.choices,
                  prob_bars=self.choices,
              )
  
          def on_train_epoch_end(self, _, pl_module):
              self.make_logging(
                  pl_module,
                  "on_train_epoch_end",
                  on_steps=[False],
                  on_epochs=[True],
                  prob_bars=self.choices,
              )
  
      class TestModel(BoringModel):
          seen_losses = []
  
          def training_step(self, batch, batch_idx):
              loss = super().training_step(batch, batch_idx)["loss"]
              self.seen_losses.append(loss)
              self.log("train_loss", loss, prog_bar=True)
              return {"loss": loss}
  
      model = TestModel()
      cb = TestCallback()
      trainer = Trainer(
          default_root_dir=tmpdir,
          limit_train_batches=2,
          limit_val_batches=0,
          num_sanity_val_steps=0,
          max_epochs=1,
          callbacks=[cb],
      )
  
      trainer.fit(model)
  
      # Make sure the func_name output equals the average from all logged values when on_epoch true
      assert (
          trainer.progress_bar_callback.get_metrics(trainer, model)[
              "train_loss"
          ]
          == model.seen_losses[-1]
      )
      assert (
          trainer.callback_metrics["train_loss"]
          == model.seen_losses[-1]
      )
  
      assert cb.call_counter == {
          "on_train_start": 1,
          "on_train_epoch_start": 1,
          "on_train_batch_start": 2,
          "on_train_batch_end": 2,
          "on_train_epoch_end": 1,
      }
  
      def get_expected(on_epoch, values):
          reduction = np.mean if on_epoch else np.max
          return reduction(values)
  
      for fx, value in trainer.callback_metrics.items():
          actual = value.item()
          if fx not in cb.logged_arguments:
              continue
          on_epoch = cb.logged_arguments[fx]["on_epoch"]
          values = cb.logged_values[fx]
          expected = get_expected(on_epoch, values)
          assert actual == expected
  
      for fx, attrs in cb.logged_arguments.items():
          should_include = (
              attrs["prog_bar"] and attrs["on_step"] ^ attrs["on_epoch"]
          )
          is_included = fx in trainer.progress_bar_metrics
          assert is_included if should_include else not is_included
  
imports: import collections
  import itertools
  from re import escape
  from unittest import mock
  from unittest.mock import call
  import numpy
  import pytest
  import torch
  from lightning.pytorch import Trainer, callbacks
  from lightning.pytorch.callbacks import EarlyStopping, ModelCheckpoint, TQDMProgressBar
  from lightning.pytorch.core.module import LightningModule
  from lightning.pytorch.demos.boring_classes import BoringModel, RandomDataset, RandomDictDataset
  from lightning.pytorch.loggers.tensorboard import TensorBoardLogger
  from lightning.pytorch.trainer.states import RunningStage
  from lightning.pytorch.utilities.exceptions import MisconfigurationException
  from lightning.pytorch.utilities.imports import _TORCHMETRICS_GREATER_EQUAL_0_11
  from lightning_utilities.test.warning import no_warning_call
  from torch import Tensor
  from torch.utils.data import DataLoader
  from torchmetrics import Accuracy
  from tests_pytorch.helpers.runif import RunIf
label_desc: Use image node_id 5 for calling a global function with example usage: test_log_works_in_train_callback(tmpdir) and returns: reduction, dict