class_docstr: CTC-attention hybrid Encoder-Decoder model
class_docstr_tok: ["CTC-attention","hybrid","Encoder-Decoder","model"]
docstr: Frontend + Encoder + Decoder + Calc loss
  
  Args:
      speech: (Batch, Length, ...)
      speech_lengths: (Batch, )
      text: (Batch, Length)
      text_lengths: (Batch,)
      kwargs: "utt_id" is among the input.
docstr_tok: ["Frontend","+","Encoder","+","Decoder","+","Calc","loss","Args",":","speech",":","(","Batch",",","Length",",","...",")","speech_lengths",":","(","Batch",",",")","text",":","(","Batch",",","Length",")","text_lengths",":","(","Batch",",",")","kwargs",":","``","utt_id","''","is","among","the","input","."]
code: def forward(
      self,
      speech: torch.Tensor,
      speech_lengths: torch.Tensor,
      text: torch.Tensor,
      text_lengths: torch.Tensor,
      **kwargs,
  ) -> Tuple[torch.Tensor, Dict[str, torch.Tensor], torch.Tensor]:
      """Frontend + Encoder + Decoder + Calc loss
  
      Args:
          speech: (Batch, Length, ...)
          speech_lengths: (Batch, )
          text: (Batch, Length)
          text_lengths: (Batch,)
          kwargs: "utt_id" is among the input.
      """
      assert text_lengths.dim() == 1, text_lengths.shape
      # Check that batch_size is unified
      assert (
          speech.shape[0]
          == speech_lengths.shape[0]
          == text.shape[0]
          == text_lengths.shape[0]
      ), (
          speech.shape,
          speech_lengths.shape,
          text.shape,
          text_lengths.shape,
      )
      batch_size = speech.shape[0]
  
      text[text == -1] = self.ignore_id
  
      # for data-parallel
      text = text[:, : text_lengths.max()]
  
      # 1. Encoder
      encoder_out, encoder_out_lens = self.encode(
          speech, speech_lengths
      )
      intermediate_outs = None
      if isinstance(encoder_out, tuple):
          intermediate_outs = encoder_out[1]
          encoder_out = encoder_out[0]
  
      loss_att, acc_att, cer_att, wer_att = None, None, None, None
      loss_ctc, cer_ctc = None, None
      loss_transducer, cer_transducer, wer_transducer = None, None, None
      stats = dict()
  
      # 1. CTC branch
      if self.ctc_weight != 0.0:
          loss_ctc, cer_ctc = self._calc_ctc_loss(
              encoder_out, encoder_out_lens, text, text_lengths
          )
  
          # Collect CTC branch stats
          stats["loss_ctc"] = (
              loss_ctc.detach() if loss_ctc is not None else None
          )
          stats["cer_ctc"] = cer_ctc
  
      # Intermediate CTC (optional)
      loss_interctc = 0.0
      if self.interctc_weight != 0.0 and intermediate_outs is not None:
          for layer_idx, intermediate_out in intermediate_outs:
              # we assume intermediate_out has the same length & padding
              # as those of encoder_out
  
              # use auxillary ctc data if specified
              loss_ic = None
              if self.aux_ctc is not None:
                  idx_key = str(layer_idx)
                  if idx_key in self.aux_ctc:
                      aux_data_key = self.aux_ctc[idx_key]
                      aux_data_tensor = kwargs.get(aux_data_key, None)
                      aux_data_lengths = kwargs.get(
                          aux_data_key + "_lengths", None
                      )
  
                      if (
                          aux_data_tensor is not None
                          and aux_data_lengths is not None
                      ):
                          loss_ic, cer_ic = self._calc_ctc_loss(
                              intermediate_out,
                              encoder_out_lens,
                              aux_data_tensor,
                              aux_data_lengths,
                          )
                      else:
                          raise Exception(
                              "Aux. CTC tasks were specified but no data was found"
                          )
              if loss_ic is None:
                  loss_ic, cer_ic = self._calc_ctc_loss(
                      intermediate_out,
                      encoder_out_lens,
                      text,
                      text_lengths,
                  )
              loss_interctc = loss_interctc + loss_ic
  
              # Collect Intermedaite CTC stats
              stats["loss_interctc_layer{}".format(layer_idx)] = (
                  loss_ic.detach() if loss_ic is not None else None
              )
              stats["cer_interctc_layer{}".format(layer_idx)] = cer_ic
  
          loss_interctc = loss_interctc / len(intermediate_outs)
  
          # calculate whole encoder loss
          loss_ctc = (
              1 - self.interctc_weight
          ) * loss_ctc + self.interctc_weight * loss_interctc
  
      if self.use_transducer_decoder:
          # 2a. Transducer decoder branch
          (
              loss_transducer,
              cer_transducer,
              wer_transducer,
          ) = self._calc_transducer_loss(
              encoder_out,
              encoder_out_lens,
              text,
          )
  
          if loss_ctc is not None:
              loss = loss_transducer + (self.ctc_weight * loss_ctc)
          else:
              loss = loss_transducer
  
          # Collect Transducer branch stats
          stats["loss_transducer"] = (
              loss_transducer.detach()
              if loss_transducer is not None
              else None
          )
          stats["cer_transducer"] = cer_transducer
          stats["wer_transducer"] = wer_transducer
  
      else:
          # 2b. Attention decoder branch
          if self.ctc_weight != 1.0:
              loss_att, acc_att, cer_att, wer_att = self._calc_att_loss(
                  encoder_out, encoder_out_lens, text, text_lengths
              )
  
          # 3. CTC-Att loss definition
          if self.ctc_weight == 0.0:
              loss = loss_att
          elif self.ctc_weight == 1.0:
              loss = loss_ctc
          else:
              loss = (
                  self.ctc_weight * loss_ctc
                  + (1 - self.ctc_weight) * loss_att
              )
  
          # Collect Attn branch stats
          stats["loss_att"] = (
              loss_att.detach() if loss_att is not None else None
          )
          stats["acc"] = acc_att
          stats["cer"] = cer_att
          stats["wer"] = wer_att
  
      # Collect total loss stats
      stats["loss"] = loss.detach()
  
      # force_gatherable: to-device and to-tensor if scalar for DataParallel
      loss, stats, weight = force_gatherable(
          (loss, stats, batch_size), loss.device
      )
      return loss, stats, weight
  
imports: import logging
  from contextlib import contextmanager
  from typing import Dict, List, Optional, Tuple, Union
  import torch
  from packaging.version import parse
  from typeguard import check_argument_types
  from espnet2.asr.ctc import CTC
  from espnet2.asr.decoder.abs_decoder import AbsDecoder
  from espnet2.asr.encoder.abs_encoder import AbsEncoder
  from espnet2.asr.frontend.abs_frontend import AbsFrontend
  from espnet2.asr.postencoder.abs_postencoder import AbsPostEncoder
  from espnet2.asr.preencoder.abs_preencoder import AbsPreEncoder
  from espnet2.asr.specaug.abs_specaug import AbsSpecAug
  from espnet2.asr.transducer.error_calculator import ErrorCalculatorTransducer
  from espnet2.asr_transducer.utils import get_transducer_task_io
  from espnet2.layers.abs_normalize import AbsNormalize
  from espnet2.torch_utils.device_funcs import force_gatherable
  from espnet2.train.abs_espnet_model import AbsESPnetModel
  from espnet.nets.e2e_asr_common import ErrorCalculator
  from espnet.nets.pytorch_backend.nets_utils import th_accuracy
  from espnet.nets.pytorch_backend.transformer.add_sos_eos import add_sos_eos
  from espnet.nets.pytorch_backend.transformer.label_smoothing_loss import LabelSmoothingLoss
label_desc: Use image node_id 2 for calling the ESPnetASRModel obj's underlying member method code with example usage: obj.forward(speech, speech_lengths, text, text_lengths) and returns: loss,  stats,  weight