class_docstr: Trainer having a optimizer.
  
  If you'd like to use multiple optimizers, then inherit this class
  and override the methods if necessary - at least "train_one_epoch()"
  
  >>> class TwoOptimizerTrainer(Trainer):
  ...     @classmethod
  ...     def add_arguments(cls, parser):
  ...         ...
  ...
  ...     @classmethod
  ...     def train_one_epoch(cls, model, optimizers, ...):
  ...         loss1 = model.model1(...)
  ...         loss1.backward()
  ...         optimizers[0].step()
  ...
  ...         loss2 = model.model2(...)
  ...         loss2.backward()
  ...         optimizers[1].step()
docstr: Perform training. This method performs the main process of training.
code: def run(
      cls,
      model: AbsESPnetModel,
      optimizers: Sequence[torch.optim.Optimizer],
      schedulers: Sequence[Optional[AbsScheduler]],
      train_iter_factory: AbsIterFactory,
      valid_iter_factory: AbsIterFactory,
      plot_attention_iter_factory: Optional[AbsIterFactory],
      trainer_options,
      distributed_option: DistributedOption,
  ) -> None:
      """Perform training. This method performs the main process of training."""
      assert check_argument_types()
      # NOTE(kamo): Don't check the type more strictly as far trainer_options
      assert is_dataclass(trainer_options), type(trainer_options)
      assert len(optimizers) == len(schedulers), (
          len(optimizers),
          len(schedulers),
      )
  
      if isinstance(trainer_options.keep_nbest_models, int):
          keep_nbest_models = [trainer_options.keep_nbest_models]
      else:
          if len(trainer_options.keep_nbest_models) == 0:
              logging.warning(
                  "No keep_nbest_models is given. Change to [1]"
              )
              trainer_options.keep_nbest_models = [1]
          keep_nbest_models = trainer_options.keep_nbest_models
  
      output_dir = Path(trainer_options.output_dir)
      reporter = Reporter()
      if trainer_options.use_amp:
          if V(torch.__version__) < V("1.6.0"):
              raise RuntimeError(
                  "Require torch>=1.6.0 for  Automatic Mixed Precision"
              )
          if trainer_options.sharded_ddp:
              if fairscale is None:
                  raise RuntimeError(
                      "Requiring fairscale. Do 'pip install fairscale'"
                  )
              scaler = fairscale.optim.grad_scaler.ShardedGradScaler()
          else:
              scaler = GradScaler()
      else:
          scaler = None
  
      use_lora = getattr(trainer_options, "use_lora", False)
      save_lora_only = getattr(trainer_options, "save_lora_only", False)
      if use_lora and lora is None:
          raise RuntimeError(
              "Requiring loralib. Do 'pip install loralib'"
          )
  
      if (
          trainer_options.resume
          and (output_dir / "checkpoint.pth").exists()
      ):
          cls.resume(
              checkpoint=output_dir / "checkpoint.pth",
              model=model,
              optimizers=optimizers,
              schedulers=schedulers,
              reporter=reporter,
              scaler=scaler,
              ngpu=trainer_options.ngpu,
              strict=not use_lora,
          )
  
      start_epoch = reporter.get_epoch() + 1
      if start_epoch == trainer_options.max_epoch + 1:
          logging.warning(
              f"The training has already reached at max_epoch: {start_epoch}"
          )
  
      if distributed_option.distributed:
          if trainer_options.sharded_ddp:
              dp_model = fairscale.nn.data_parallel.ShardedDataParallel(
                  module=model,
                  sharded_optimizer=optimizers,
              )
          else:
              dp_model = torch.nn.parallel.DistributedDataParallel(
                  model,
                  device_ids=(
                      # Perform multi-Process with multi-GPUs
                      [torch.cuda.current_device()]
                      if distributed_option.ngpu == 1
                      # Perform single-Process with multi-GPUs
                      else None
                  ),
                  output_device=(
                      torch.cuda.current_device()
                      if distributed_option.ngpu == 1
                      else None
                  ),
                  find_unused_parameters=trainer_options.unused_parameters,
              )
      elif distributed_option.ngpu > 1:
          dp_model = torch.nn.parallel.DataParallel(
              model,
              device_ids=list(range(distributed_option.ngpu)),
          )
      else:
          # NOTE(kamo): DataParallel also should work with ngpu=1,
          # but for debuggability it's better to keep this block.
          dp_model = model
  
      if trainer_options.use_tensorboard and (
          not distributed_option.distributed
          or distributed_option.dist_rank == 0
      ):
          from torch.utils.tensorboard import SummaryWriter
  
          train_summary_writer = SummaryWriter(
              str(output_dir / "tensorboard" / "train")
          )
          valid_summary_writer = SummaryWriter(
              str(output_dir / "tensorboard" / "valid")
          )
      else:
          train_summary_writer = None
  
      start_time = time.perf_counter()
      for iepoch in range(start_epoch, trainer_options.max_epoch + 1):
          if iepoch != start_epoch:
              logging.info(
                  "{}/{}epoch started. Estimated time to finish: {}".format(
                      iepoch,
                      trainer_options.max_epoch,
                      humanfriendly.format_timespan(
                          (time.perf_counter() - start_time)
                          / (iepoch - start_epoch)
                          * (trainer_options.max_epoch - iepoch + 1)
                      ),
                  )
              )
          else:
              logging.info(
                  f"{iepoch}/{trainer_options.max_epoch}epoch started"
              )
          set_all_random_seed(trainer_options.seed + iepoch)
  
          reporter.set_epoch(iepoch)
          # 1. Train and validation for one-epoch
          with reporter.observe("train") as sub_reporter:
              all_steps_are_invalid = cls.train_one_epoch(
                  model=dp_model,
                  optimizers=optimizers,
                  schedulers=schedulers,
                  iterator=train_iter_factory.build_iter(iepoch),
                  reporter=sub_reporter,
                  scaler=scaler,
                  summary_writer=train_summary_writer,
                  options=trainer_options,
                  distributed_option=distributed_option,
              )
  
          with reporter.observe("valid") as sub_reporter:
              cls.validate_one_epoch(
                  model=dp_model,
                  iterator=valid_iter_factory.build_iter(iepoch),
                  reporter=sub_reporter,
                  options=trainer_options,
                  distributed_option=distributed_option,
              )
          if (
              not distributed_option.distributed
              or distributed_option.dist_rank == 0
          ):
              # att_plot doesn't support distributed
              if plot_attention_iter_factory is not None:
                  with reporter.observe("att_plot") as sub_reporter:
                      cls.plot_attention(
                          model=model,
                          output_dir=output_dir / "att_ws",
                          summary_writer=train_summary_writer,
                          iterator=plot_attention_iter_factory.build_iter(
                              iepoch
                          ),
                          reporter=sub_reporter,
                          options=trainer_options,
                      )
  
          # 2. LR Scheduler step
          for scheduler in schedulers:
              if isinstance(scheduler, AbsValEpochStepScheduler):
                  scheduler.step(
                      reporter.get_value(
                          *trainer_options.val_scheduler_criterion
                      )
                  )
              elif isinstance(scheduler, AbsEpochStepScheduler):
                  scheduler.step()
          if trainer_options.sharded_ddp:
              for optimizer in optimizers:
                  if isinstance(optimizer, fairscale.optim.oss.OSS):
                      optimizer.consolidate_state_dict()
  
          if (
              not distributed_option.distributed
              or distributed_option.dist_rank == 0
          ):
              # 3. Report the results
              logging.info(reporter.log_message())
              if trainer_options.use_matplotlib:
                  reporter.matplotlib_plot(output_dir / "images")
              if train_summary_writer is not None:
                  reporter.tensorboard_add_scalar(
                      train_summary_writer, key1="train"
                  )
                  reporter.tensorboard_add_scalar(
                      valid_summary_writer, key1="valid"
                  )
              if trainer_options.use_wandb:
                  reporter.wandb_log()
  
              # 4. Save/Update the checkpoint
              if use_lora and save_lora_only:
                  # Only the LoRA realted params are saved, not the whole model
                  model_state_dict = lora.lora_state_dict(model)
              else:
                  # Save all params of the model
                  model_state_dict = model.state_dict()
  
              torch.save(
                  {
                      "model": model_state_dict,
                      "reporter": reporter.state_dict(),
                      "optimizers": [
                          o.state_dict() for o in optimizers
                      ],
                      "schedulers": [
                          s.state_dict() if s is not None else None
                          for s in schedulers
                      ],
                      "scaler": scaler.state_dict()
                      if scaler is not None
                      else None,
                  },
                  output_dir / "checkpoint.pth",
              )
  
              # 5. Save and log the model and update the link to the best model
              torch.save(
                  model_state_dict, output_dir / f"{iepoch}epoch.pth"
              )
  
              # Creates a sym link latest.pth -> {iepoch}epoch.pth
              p = output_dir / "latest.pth"
              if p.is_symlink() or p.exists():
                  p.unlink()
              p.symlink_to(f"{iepoch}epoch.pth")
  
              _improved = []
              for (
                  _phase,
                  k,
                  _mode,
              ) in trainer_options.best_model_criterion:
                  # e.g. _phase, k, _mode = "train", "loss", "min"
                  if reporter.has(_phase, k):
                      best_epoch = reporter.get_best_epoch(
                          _phase, k, _mode
                      )
                      # Creates sym links if it's the best result
                      if best_epoch == iepoch:
                          p = output_dir / f"{_phase}.{k}.best.pth"
                          if p.is_symlink() or p.exists():
                              p.unlink()
                          p.symlink_to(f"{iepoch}epoch.pth")
                          _improved.append(f"{_phase}.{k}")
              if len(_improved) == 0:
                  logging.info(
                      "There are no improvements in this epoch"
                  )
              else:
                  logging.info(
                      "The best model has been updated: "
                      + ", ".join(_improved)
                  )
  
              log_model = (
                  trainer_options.wandb_model_log_interval > 0
                  and iepoch % trainer_options.wandb_model_log_interval
                  == 0
              )
              if log_model and trainer_options.use_wandb:
                  import wandb
  
                  logging.info("Logging Model on this epoch :::::")
                  artifact = wandb.Artifact(
                      name=f"model_{wandb.run.id}",
                      type="model",
                      metadata={"improved": _improved},
                  )
                  artifact.add_file(
                      str(output_dir / f"{iepoch}epoch.pth")
                  )
                  aliases = [
                      f"epoch-{iepoch}",
                      "best" if best_epoch == iepoch else "",
                  ]
                  wandb.log_artifact(artifact, aliases=aliases)
  
              # 6. Remove the model files excluding n-best epoch and latest epoch
              _removed = []
              # Get the union set of the n-best among multiple criterion
              nbests = set().union(
                  *[
                      set(
                          reporter.sort_epochs(ph, k, m)[
                              : max(keep_nbest_models)
                          ]
                      )
                      for ph, k, m in trainer_options.best_model_criterion
                      if reporter.has(ph, k)
                  ]
              )
  
              # Generated n-best averaged model
              if (
                  trainer_options.nbest_averaging_interval > 0
                  and iepoch % trainer_options.nbest_averaging_interval
                  == 0
              ):
                  average_nbest_models(
                      reporter=reporter,
                      output_dir=output_dir,
                      best_model_criterion=trainer_options.best_model_criterion,
                      nbest=keep_nbest_models,
                      suffix=f"till{iepoch}epoch",
                  )
  
              for e in range(1, iepoch):
                  p = output_dir / f"{e}epoch.pth"
                  if p.exists() and e not in nbests:
                      p.unlink()
                      _removed.append(str(p))
              if len(_removed) != 0:
                  logging.info(
                      "The model files were removed: "
                      + ", ".join(_removed)
                  )
  
          # 7. If any updating haven't happened, stops the training
          if all_steps_are_invalid:
              logging.warning(
                  "The gradients at all steps are invalid in this epoch. "
                  f"Something seems wrong. This training was stopped at {iepoch}epoch"
              )
              break
  
          # 8. Check early stopping
          if trainer_options.patience is not None:
              if reporter.check_early_stopping(
                  trainer_options.patience,
                  *trainer_options.early_stopping_criterion,
              ):
                  break
  
      else:
          logging.info(
              f"The training was finished at {trainer_options.max_epoch} epochs "
          )
  
      # Generated n-best averaged model
      if (
          not distributed_option.distributed
          or distributed_option.dist_rank == 0
      ):
          average_nbest_models(
              reporter=reporter,
              output_dir=output_dir,
              best_model_criterion=trainer_options.best_model_criterion,
              nbest=keep_nbest_models,
          )
  
imports: import argparse
  import dataclasses
  import logging
  import time
  from contextlib import contextmanager
  from dataclasses import is_dataclass
  from pathlib import Path
  from typing import Dict, Iterable, List, Optional, Sequence, Tuple, Union
  import humanfriendly
  import numpy
  import torch
  import torch.nn
  import torch.optim
  from packaging.version import parse
  from typeguard import check_argument_types
  from espnet2.iterators.abs_iter_factory import AbsIterFactory
  from espnet2.main_funcs.average_nbest_models import average_nbest_models
  from espnet2.main_funcs.calculate_all_attentions import calculate_all_attentions
  from espnet2.schedulers.abs_scheduler import AbsBatchStepScheduler, AbsEpochStepScheduler, AbsScheduler, AbsValEpochStepScheduler
  from espnet2.torch_utils.add_gradient_noise import add_gradient_noise
  from espnet2.torch_utils.device_funcs import to_device
  from espnet2.torch_utils.recursive_op import recursive_average
  from espnet2.torch_utils.set_all_random_seed import set_all_random_seed
  from espnet2.train.abs_espnet_model import AbsESPnetModel
  from espnet2.train.distributed_utils import DistributedOption
  from espnet2.train.reporter import Reporter, SubReporter
  from espnet2.utils.build_dataclass import build_dataclass
  from espnet2.utils.kwargs2args import kwargs2args
label_desc: Use image node_id 5 for calling the Trainer obj's underlying member method code with example usage: obj.run(cls, model, optimizers, schedulers, train_iter_factory, valid_iter_factory, plot_attention_iter_factory, trainer_options, distributed_option) without return types