code: def on_log(
      self,
      args: TrainingArguments,
      state: TrainerState,
      control: TrainerControl,
      logs: Optional[Dict[str, Any]] = None,
      **kwargs: Any,
  ) -> None:
      if logs is None:
          logger.warning("on_log called with empty logs")
          return
      metrics, metric_type = self._get_metrics(logs)
      logger.debug(
          f"on_log metrics, global_step {state.global_step}", metrics
      )
      if metric_type == TRAIN:
          # Prevents reporting metrics for the same step twice. This happens after
          # training is completed and average training metrics are reported with
          # the same step as the in-progress training metrics.
          if self.last_metrics["train_step"] != state.global_step:
              if state.is_world_process_zero:
                  self.core_context.train.report_training_metrics(
                      steps_completed=state.global_step, metrics=metrics
                  )
              metrics["train_step"] = state.global_step
  
      elif metric_type == EVAL:
          # Prevents reporting metrics for the same step twice. This happens when
          # after-training evaluation is completed, and it is reported with the same
          # step as the last during-training evaluation.
          if self.last_metrics["eval_step"] != state.global_step:
              if state.is_world_process_zero:
                  self.core_context.train.report_validation_metrics(
                      steps_completed=state.global_step, metrics=metrics
                  )
              metrics["eval_step"] = state.global_step
      else:
          logger.warning(
              f"Metrics not reported: metric type = {metric_type}."
          )
  
      self.last_metrics.update(metrics)
  
      # Update searcher state after collecting the metrics.
      if self.updating_searcher is True:
          self._update_searcher(state, control)
  
      # If searcher is NOT being updated and preemption signal is received
      # (e.g., by pausing experiment in the WebUI), notify Trainer (via TrainerControl)
      # to save the checkpoint. After the checkpoint is uploaded to Determined storage,
      # the process is preempted (see on_save() method for details).
      if (
          self.updating_searcher is False
          and self.core_context.preempt.should_preempt()
      ):
          control.should_save = True
  
imports: import json
  import logging
  import os
  from typing import Any, Dict, List, Optional, Tuple
  from transformers import TrainerCallback, TrainerControl, TrainerState, TrainingArguments
  from transformers.trainer_utils import get_last_checkpoint
  import determined
label_desc: Use image node_id 2 for calling the DetCallback obj's underlying member method code with example usage: obj.on_log(args, state, control, logs) without return types