class_docstr: Unsupervised ASR model.
  
  The source code is from FAIRSEQ:
  https://github.com/facebookresearch/fairseq/tree/main/examples/wav2vec/unsupervised
docstr: Frontend + Segmenter + Generator + Discriminator + Calc Loss
  
  Args:
docstr_tok: ["Frontend","+","Segmenter","+","Generator","+","Discriminator","+","Calc","Loss","Args",":"]
code: def forward(
      self,
      speech: torch.Tensor,
      speech_lengths: torch.Tensor,
      text: Optional[torch.Tensor] = None,
      text_lengths: Optional[torch.Tensor] = None,
      pseudo_labels: Optional[torch.Tensor] = None,
      pseudo_labels_lengths: Optional[torch.Tensor] = None,
      do_validation: Optional[str2bool] = False,
      print_hyp: Optional[str2bool] = False,
      **kwargs,
  ) -> Tuple[torch.Tensor, Dict[str, torch.Tensor], torch.Tensor]:
      """Frontend + Segmenter + Generator + Discriminator + Calc Loss
  
      Args:
      """
      stats = {}
  
      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]
  
      # for data-parallel
      text = text[:, : text_lengths.max()]
  
      # 1. Feats encode (Extract feats + Apply segmenter)
      feats, padding_mask = self.encode(speech, speech_lengths)
  
      # 2. Generate fake samples
      (
          generated_sample,
          real_sample,
          x_inter,
          generated_sample_padding_mask,
      ) = self.generator(feats, text, padding_mask)
  
      # 3. Reprocess segments
      if self.use_segmenter:
          (
              generated_sample,
              generated_sample_padding_mask,
          ) = self.segmenter.logit_segment(
              generated_sample, generated_sample_padding_mask
          )
  
      # for phone_diversity_loss
      generated_sample_logits = generated_sample
  
      if not self.skip_softmax:
          if self.training and self.use_gumbel:
              generated_sample = F.gumbel_softmax(
                  generated_sample_logits.float(),
                  tau=self.curr_temp,
                  hard=self.use_hard_gumbel,
              ).type_as(generated_sample_logits)
          else:
              generated_sample = generated_sample_logits.softmax(-1)
  
      # for validation
      vocab_seen = None
      if do_validation:
          batch_num_errors = 0
          batched_hyp_ids = generated_sample.argmax(-1)
          batched_hyp_ids[generated_sample_padding_mask] = self.pad
  
          # for kenlm ppl metric
          batch_lm_log_prob = 0
          batch_num_hyp_tokens = 0
          vocab_seen = torch.zeros(
              self.vocab_size - 4, dtype=torch.bool
          )
  
          for hyp_ids, ref_ids in zip(batched_hyp_ids, text):
              # remove <pad> and <unk>
              hyp_ids = hyp_ids[hyp_ids >= 4]
              # remove duplicate tokens
              hyp_ids = hyp_ids.unique_consecutive()
              # remove silence
              hyp_ids_nosil = hyp_ids[hyp_ids != self.sil]
              hyp_ids_nosil_list = hyp_ids_nosil.tolist()
  
              if self.kenlm:
                  hyp_token_list = self.token_id_converter.ids2tokens(
                      integers=hyp_ids
                  )
                  hyp_tokens = " ".join(hyp_token_list)
                  lm_log_prob = self.kenlm.score(hyp_tokens)
                  batch_lm_log_prob += lm_log_prob
                  batch_num_hyp_tokens += len(hyp_token_list)
  
                  hyp_tokens_index = hyp_ids[hyp_ids >= 4]
                  vocab_seen[hyp_tokens_index - 4] = True
  
              ref_ids = ref_ids[ref_ids != self.pad]
              ref_ids_list = ref_ids.tolist()
              num_errors = editdistance.eval(
                  hyp_ids_nosil_list, ref_ids_list
              )
              batch_num_errors += num_errors
  
          stats["batch_num_errors"] = batch_num_errors
          stats["batch_num_ref_tokens"] = text_lengths.sum().item()
          if self.kenlm:
              stats["batch_lm_log_prob"] = batch_lm_log_prob
              stats["batch_num_hyp_tokens"] = batch_num_hyp_tokens
              stats["batch_size"] = batch_size
  
          # print the last sample in the batch
          if print_hyp:
              hyp_token_list = self.token_id_converter.ids2tokens(
                  integers=hyp_ids_nosil
              )
              hyp_tokens = " ".join(hyp_token_list)
  
              ref_token_list = self.token_id_converter.ids2tokens(
                  integers=ref_ids
              )
              ref_tokens = " ".join(ref_token_list)
              logging.info(f"[REF]: {ref_tokens}")
              logging.info(f"[HYP]: {hyp_tokens}")
  
      real_sample_padding_mask = text == self.pad
      # 5. Discriminator condition
      generated_sample_prediction = self.discriminator(
          generated_sample, generated_sample_padding_mask
      )
      real_sample_prediction = self.discriminator(
          real_sample, real_sample_padding_mask
      )
  
      is_discriminative_step = self.is_discriminative_step()
  
      # 5. Calculate losses
      loss_info = []
  
      if "discriminator_loss" in self.losses.keys():
          (
              generated_sample_prediction_loss,
              real_sample_prediction_loss,
          ) = self.losses["discriminator_loss"](
              generated_sample_prediction,
              real_sample_prediction,
              is_discriminative_step,
          )
          loss_info.append(
              generated_sample_prediction_loss
              * self.losses["discriminator_loss"].weight
          )
          if is_discriminative_step:
              loss_info.append(
                  real_sample_prediction_loss
                  * self.losses["discriminator_loss"].weight
              )
      else:
          (
              generated_sample_prediction_loss,
              real_sample_prediction_loss,
          ) = (None, None)
  
      if "gradient_penalty" in self.losses.keys():
          gp = self.losses["gradient_penalty"](
              generated_sample,
              real_sample,
              self.training,
              is_discriminative_step,
          )
          loss_info.append(gp * self.losses["gradient_penalty"].weight)
          loss_info.append(gp * self.losses["gradient_penalty"].weight)
      else:
          gp = None
  
      if "phoneme_diversity_loss" in self.losses.keys():
          pdl = self.losses["phoneme_diversity_loss"](
              generated_sample_logits,
              batch_size,
              is_discriminative_step,
          )
          loss_info.append(
              pdl * self.losses["phoneme_diversity_loss"].weight
          )
      else:
          pdl = None
  
      if "smoothness_penalty" in self.losses.keys():
          sp = self.losses["smoothness_penalty"](
              generated_sample_logits,
              generated_sample_padding_mask,
              batch_size,
              is_discriminative_step,
          )
          loss_info.append(
              sp * self.losses["smoothness_penalty"].weight
          )
      else:
          sp = None
  
      if (
          "pseudo_label_loss" in self.losses.keys()
          and pseudo_labels is not None
      ):
          mmi = self.losses["pseudo_label_loss"](
              x_inter, pseudo_labels, is_discriminative_step
          )
          loss_info.append(
              mmi * self.losses["pseudo_label_loss"].weight
          )
      else:
          mmi = None
  
      # Update temperature
      self._change_temperature()
      self.number_updates += 1
  
      loss = sum(loss_info)
  
      # Collect total loss stats
      stats["loss"] = loss.detach()
      stats[
          "generated_sample_prediction_loss"
      ] = generated_sample_prediction_loss
      stats["real_sample_prediction_loss"] = real_sample_prediction_loss
      stats["gp"] = gp
      stats["sp"] = sp
      stats["pdl"] = pdl
      stats["mmi"] = mmi
  
      # 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, vocab_seen
  
imports: import argparse
  import logging
  from contextlib import contextmanager
  from typing import Dict, Optional, Tuple
  import editdistance
  import torch
  import torch.nn.functional
  from packaging.version import parse
  from typeguard import check_argument_types
  from espnet2.asr.frontend.abs_frontend import AbsFrontend
  from espnet2.text.token_id_converter import TokenIDConverter
  from espnet2.torch_utils.device_funcs import force_gatherable
  from espnet2.train.abs_espnet_model import AbsESPnetModel
  from espnet2.uasr.discriminator.abs_discriminator import AbsDiscriminator
  from espnet2.uasr.generator.abs_generator import AbsGenerator
  from espnet2.uasr.loss.abs_loss import AbsUASRLoss
  from espnet2.uasr.segmenter.abs_segmenter import AbsSegmenter
  from espnet2.utils.types import str2bool
  from espnet.nets.pytorch_backend.nets_utils import make_pad_mask
label_desc: Use image node_id 4 for calling the ESPnetUASRModel obj's underlying member method code with example usage: obj.forward(speech, speech_lengths, text, text_lengths, pseudo_labels, pseudo_labels_lengths, do_validation, print_hyp) and returns: loss,  stats,  weight,  vocab_seen