code: def train(
      self,
      training_set: Union["Dataset", "RayDataset"],  # noqa: F821
      validation_set: Optional[
          Union["Dataset", "RayDataset"]
      ],  # noqa: F821
      test_set: Optional[Union["Dataset", "RayDataset"]],  # noqa: F821
      save_path="model",
      **kwargs,
  ):
      # ====== General setup =======
      output_features = self.model.output_features
  
      # Only use signals when on the main thread to avoid issues with CherryPy
      # https://github.com/ludwig-ai/ludwig/issues/286
      if threading.current_thread() == threading.main_thread():
          # set the original sigint signal handler
          # as we want to restore it at the end of training
          self.original_sigint_handler = signal.getsignal(signal.SIGINT)
          signal.signal(signal.SIGINT, self.set_steps_to_1_or_quit)
  
      # TODO: construct new datasets by running encoders (for text, image)
  
      # ====== Setup file names =======
      training_checkpoints_path = None
      tensorboard_log_dir = None
      if self.is_coordinator():
          os.makedirs(save_path, exist_ok=True)
          training_checkpoints_path = os.path.join(
              save_path, TRAINING_CHECKPOINTS_DIR_PATH
          )
          tensorboard_log_dir = os.path.join(save_path, "logs")
  
      self.callback(
          lambda c: c.on_trainer_train_setup(
              self, save_path, self.is_coordinator()
          ),
          coordinator_only=False,
      )
  
      # ====== Setup session =======
      checkpoint = checkpoint_manager = None
      if self.is_coordinator() and not self.skip_save_progress:
          checkpoint = self.distributed.create_checkpoint_handle(
              dist_model=self.model, model=self.model
          )
          checkpoint_manager = CheckpointManager(
              checkpoint, training_checkpoints_path, device=self.device
          )
  
      train_summary_writer = None
      validation_summary_writer = None
      test_summary_writer = None
      if (
          self.is_coordinator()
          and not self.skip_save_log
          and tensorboard_log_dir
      ):
          train_summary_writer = SummaryWriter(
              os.path.join(tensorboard_log_dir, TRAINING)
          )
          if validation_set is not None and validation_set.size > 0:
              validation_summary_writer = SummaryWriter(
                  os.path.join(tensorboard_log_dir, VALIDATION)
              )
          if test_set is not None and test_set.size > 0:
              test_summary_writer = SummaryWriter(
                  os.path.join(tensorboard_log_dir, TEST)
              )
  
      progress_tracker = get_new_progress_tracker(
          batch_size=-1,
          learning_rate=self.base_learning_rate,
          best_eval_metric_value=get_initial_validation_value(
              self.validation_metric
          ),
          best_increase_batch_size_eval_metric=float("inf"),
          output_features=output_features,
      )
  
      set_random_seed(self.random_seed)
  
      try:
          params = self._construct_lgb_params()
  
          (
              lgb_train,
              eval_sets,
              eval_names,
          ) = self._construct_lgb_datasets(
              training_set, validation_set, test_set
          )
  
          # use separate total steps variable to allow custom SIGINT logic
          self.total_steps = self.num_boost_round
  
          early_stopping_steps = (
              self.boosting_rounds_per_checkpoint * self.early_stop
          )
  
          if self.is_coordinator():
              logger.info(
                  f"Training for {self.total_steps} boosting round(s), approximately "
                  f"{int(self.total_steps / self.boosting_rounds_per_checkpoint)} round(s) of evaluation."
              )
              logger.info(
                  f"Early stopping policy: {self.early_stop} round(s) of evaluation, or {early_stopping_steps} "
                  f"boosting round(s).\n"
              )
  
              logger.info(
                  f"Starting with step {progress_tracker.steps}"
              )
  
          progress_bar_config = {
              "desc": "Training",
              "total": self.total_steps,
              "disable": is_progressbar_disabled(),
              "file": sys.stdout,
          }
          progress_bar = LudwigProgressBar(
              self.report_tqdm_to_ray,
              progress_bar_config,
              self.is_coordinator(),
          )
  
          while progress_tracker.steps < self.total_steps:
              # epoch init
              start_time = time.time()
  
              # Reset the metrics at the start of the next epoch
              self.model.reset_metrics()
  
              self.callback(
                  lambda c: c.on_epoch_start(
                      self, progress_tracker, save_path
                  )
              )
  
              should_break = self._train_loop(
                  params,
                  lgb_train,
                  eval_sets,
                  eval_names,
                  progress_tracker,
                  progress_bar,
                  save_path,
                  training_set,
                  validation_set,
                  test_set,
                  train_summary_writer,
                  validation_summary_writer,
                  test_summary_writer,
                  early_stopping_steps,
              )
  
              # ================ Post Training Epoch ================
              progress_tracker.epoch += 1
              self.callback(
                  lambda c: c.on_epoch_end(
                      self, progress_tracker, save_path
                  )
              )
  
              if self.is_coordinator():
                  # ========== Save training progress ==========
                  logger.debug(
                      f"Epoch {progress_tracker.epoch} took: "
                      f"{time_utils.strdelta((time.time()- start_time) * 1000.0)}."
                  )
                  if not self.skip_save_progress:
                      progress_tracker.checkpoint_number += 1
                      checkpoint_manager.checkpoint.model = self.model
                      checkpoint_manager.save(progress_tracker.steps)
                      progress_tracker.save(
                          os.path.join(
                              save_path,
                              TRAINING_PROGRESS_TRACKER_FILE_NAME,
                          )
                      )
  
                      self.callback(
                          lambda c: c.on_checkpoint(
                              self, progress_tracker
                          )
                      )
  
              # Early stop if needed.
              if should_break:
                  break
      finally:
          # ================ Finished Training ================
          if (
              not self.model.has_saved(save_path)
              and self.is_coordinator()
              and not self.skip_save_model
          ):
              try:
                  self.model.save(save_path)
              except ValueError:
                  logger.info(
                      "No LightGBM model initialized, skipping save"
                  )
  
          self.callback(
              lambda c: c.on_trainer_train_teardown(
                  self,
                  progress_tracker,
                  save_path,
                  self.is_coordinator(),
              ),
              coordinator_only=False,
          )
  
          if train_summary_writer is not None:
              train_summary_writer.close()
          if validation_summary_writer is not None:
              validation_summary_writer.close()
          if test_summary_writer is not None:
              test_summary_writer.close()
  
          if self.is_coordinator() and not self.skip_save_progress:
              checkpoint_manager.close()
  
      # Load the best weights from saved checkpoint
      if self.is_coordinator() and not self.skip_save_model:
          self.model.load(save_path)
  
      # restore original sigint signal handler
      if (
          self.original_sigint_handler
          and threading.current_thread() == threading.main_thread()
      ):
          signal.signal(signal.SIGINT, self.original_sigint_handler)
  
      return (
          self.model,
          progress_tracker.train_metrics,
          progress_tracker.validation_metrics,
          progress_tracker.test_metrics,
      )
  
imports: import logging
  import os
  import re
  import signal
  import sys
  import threading
  import time
  from typing import Any, Dict, List, Optional, Tuple, Union
  import lightgbm
  import torch
  from torch.utils.tensorboard import SummaryWriter
  from ludwig.constants import BINARY, CATEGORY, MINIMIZE, MODEL_GBM, NUMBER, TEST, TRAINING, VALIDATION
  from ludwig.distributed import init_dist_strategy
  from ludwig.distributed.base import DistributedStrategy, LocalStrategy
  from ludwig.features.feature_utils import LudwigFeatureDict
  from ludwig.globals import is_progressbar_disabled, TRAINING_CHECKPOINTS_DIR_PATH, TRAINING_PROGRESS_TRACKER_FILE_NAME
  from ludwig.models.gbm import GBM
  from ludwig.models.predictor import Predictor
  from ludwig.modules.metric_modules import get_improved_fn, get_initial_validation_value
  from ludwig.modules.metric_registry import get_metric_objective
  from ludwig.progress_bar import LudwigProgressBar
  from ludwig.schema.trainer import BaseTrainerConfig, GBMTrainerConfig
  from ludwig.trainers.base import BaseTrainer
  from ludwig.trainers.registry import register_ray_trainer, register_trainer
  from ludwig.types import ModelConfigDict
  from ludwig.utils import time_utils
  from ludwig.utils.checkpoint_utils import CheckpointManager
  from ludwig.utils.defaults import default_random_seed
  from ludwig.utils.gbm_utils import get_single_output_feature, get_targets, log_loss_objective, logits_to_predictions, multiclass_objective, store_predictions, store_predictions_ray, TrainLogits
  from ludwig.utils.metric_utils import get_metric_names, TrainerMetric
  from ludwig.utils.metrics_printed_table import print_metrics_table
  from ludwig.utils.misc_utils import set_random_seed
  from ludwig.utils.trainer_utils import append_metrics, get_latest_metrics_dict, get_new_progress_tracker, ProgressTracker
label_desc: Use image node_id 13 for calling the LightGBMTrainer obj's underlying member method code with example usage: obj.train(training_set, validation_set, test_set, save_path) and returns: self,  progress_tracker,  progress_tracker,  progress_tracker