class_docstr: Speech2Speech class.
docstr: Run speech-to-speech.
code: def __call__(
      self,
      src_speech: Union[torch.Tensor, np.ndarray],
      src_speech_lengths: Union[torch.Tensor, np.ndarray] = None,
      tgt_speech: Union[torch.Tensor, np.ndarray] = None,
      tgt_speech_lengths: Union[torch.Tensor, np.ndarray] = None,
      spembs: Union[torch.Tensor, np.ndarray] = None,
      sids: Union[torch.Tensor, np.ndarray] = None,
      lids: Union[torch.Tensor, np.ndarray] = None,
      decode_conf: Optional[Dict[str, Any]] = None,
  ) -> Dict[str, torch.Tensor]:
      """Run speech-to-speech."""
      assert check_argument_types()
  
      # check inputs
      if self.use_speech and tgt_speech is None:
          raise RuntimeError("Missing required argument: 'tgt_speech'")
      if self.use_sids and sids is None:
          raise RuntimeError("Missing required argument: 'sids'")
      if self.use_lids and lids is None:
          raise RuntimeError("Missing required argument: 'lids'")
      if self.use_spembs and spembs is None:
          raise RuntimeError("Missing required argument: 'spembs'")
  
      # Input as audio signal
      if isinstance(src_speech, np.ndarray):
          src_speech = torch.tensor(src_speech.astype(np.float32))
  
      if src_speech_lengths is None:
          src_speech_lengths = src_speech.new_full(
              [1], dtype=torch.long, fill_value=src_speech.size(1)
          )
  
      # prepare batch
      batch = dict(
          src_speech=src_speech, src_speech_lengths=src_speech_lengths
      )
      if tgt_speech is not None:
          batch.update(tgt_speech=tgt_speech)
          batch.update(tgt_speech_lengths=tgt_speech_lengths)
      if spembs is not None:
          batch.update(spembs=spembs)
      if sids is not None:
          batch.update(sids=sids)
      if lids is not None:
          batch.update(lids=lids)
      batch = to_device(batch, self.device)
  
      # overwrite the decode configs if provided
      cfg = self.decode_conf
      if decode_conf is not None:
          cfg = self.decode_conf.copy()
          cfg.update(decode_conf)
  
      # inference
      if self.always_fix_seed:
          set_all_random_seed(self.seed)
  
      if self.s2st_type == "translatotron":
          output_dict = self.model.inference(**batch, **cfg)
          # apply vocoder (mel-to-wav)
          if self.vocoder is not None:
              if (
                  self.prefer_normalized_feats
                  or output_dict.get("feat_gen_denorm") is None
              ):
                  input_feat = output_dict["feat_gen"]
              else:
                  input_feat = output_dict["feat_gen_denorm"]
              wav = self.vocoder(input_feat)
              output_dict.update(wav=wav)
      elif self.s2st_type == "discrete_unit":
          output_dict = {}
          # Forward Encoder
          enc, _ = self.model.encode(
              batch["src_speech"], batch["src_speech_lengths"]
          )
          nbest_hyps = self.beam_search(
              x=enc[0],
              maxlenratio=self.maxlenratio,
              minlenratio=self.minlenratio,
          )
          # TODO(jiatong): get nbest list instead of just best hyp
          best_hyp = nbest_hyps[0]  # just use the best
          # remove sos/eos and get results
          token_int = np.array(best_hyp.yseq[1:-1].tolist())
          output_dict.update(feat_gen=torch.tensor(token_int))
  
          logging.info("token_int: {}".format(token_int))
  
          if self.vocoder is not None:
              if len(token_int) == 0:
                  output_dict.update(wav=torch.tensor([0] * 100))
              else:
                  input_discrete_unit = to_device(
                      torch.tensor(token_int).view(-1, 1),
                      device=self.device,
                  )
                  # NOTE(jiatong): we default take the last token
                  # in the token list as <unk>
                  # see scripts/feats/performa_kemans.sh for details
                  input_discrete_unit = input_discrete_unit[
                      input_discrete_unit
                      != self.model.unit_vocab_size - 1
                  ].view(-1, 1)
                  wav = self.vocoder(input_discrete_unit)
                  output_dict.update(wav=wav)
  
      elif self.s2st_type == "unity":
          output_dict = {}
          # Forward Encoder
          enc, _ = self.model.encode(
              batch["src_speech"], batch["src_speech_lengths"]
          )
  
          st_subtask_nbest_hyps = self.st_subtask_beam_search(
              x=enc[0],
              maxlenratio=self.st_subtask_maxlenratio,
              minlenratio=self.st_subtask_minlenratio,
          )
  
          logging.info(
              "st_subtask_token_int: {}".format(
                  st_subtask_nbest_hyps[0].yseq[1:-1].tolist()
              )
          )
  
          st_subtask_result = []
          for hyp in st_subtask_nbest_hyps:
              assert isinstance(hyp, Hypothesis), type(hyp)
  
              # remove sos/eos and get results
              if isinstance(hyp.hs, List):
                  st_subtask_hs = torch.stack(hyp.hs)
              else:
                  st_subtask_hs = hyp.hs
  
              st_subtask_token_int = hyp.yseq[1:-1].tolist()
              st_subtask_token = self.st_subtask_converter.ids2tokens(
                  st_subtask_token_int
              )
              if self.st_subtask_tokenizer is not None:
                  st_subtask_hyp_text = (
                      self.st_subtask_tokenizer.tokens2text(
                          st_subtask_token
                      )
                  )
              else:
                  st_subtask_hyp_text = None
  
              st_subtask_result.append(
                  (
                      st_subtask_hyp_text,
                      st_subtask_token,
                      st_subtask_token_int,
                      st_subtask_hs,
                  )
              )
  
          if self.st_subtask_tokenizer is not None:
              (
                  st_subtask_hyp_text,
                  st_subtask_token,
                  st_subtask_token_int,
                  _,
              ) = st_subtask_result[0]
              logging.info(
                  "st_subtask_text: {}".format(st_subtask_result[0][0])
              )
              output_dict.update(st_subtask_text=st_subtask_hyp_text)
              output_dict.update(st_subtask_token=st_subtask_token)
              output_dict.update(
                  st_subtask_token_int=st_subtask_token_int
              )
  
          # encoder 1best st_subtask result
          st_subtask_hs = st_subtask_result[0][-1].unsqueeze(0)
          st_subtask_hs = to_device(st_subtask_hs, device=self.device)
          st_subtask_hs_lengths = st_subtask_hs.new_full(
              [1], dtype=torch.long, fill_value=st_subtask_hs.size(1)
          )
          md_enc, _, _ = self.model.unit_encoder(
              st_subtask_hs, st_subtask_hs_lengths
          )
          nbest_hyps = self.beam_search(
              md_enc[0],
              maxlenratio=self.maxlenratio * 100,
              minlenratio=self.minlenratio,
          )
  
          # TODO(jiatong): get nbest list instead of just best hyp
          best_hyp = nbest_hyps[0]  # just use the best
          # remove sos/eos and get results
          token_int = np.array(best_hyp.yseq[1:-1].tolist())
          output_dict.update(feat_gen=torch.tensor(token_int))
  
          logging.info("token_int: {}".format(token_int))
  
          if self.vocoder is not None:
              if len(token_int) == 0:
                  output_dict.update(wav=torch.tensor([0] * 100))
              else:
                  input_discrete_unit = to_device(
                      torch.tensor(token_int).view(-1, 1),
                      device=self.device,
                  )
                  # NOTE(jiatong): we default take the last token
                  # in the token list as <unk>
                  # see scripts/feats/performa_kemans.sh for details
                  input_discrete_unit = input_discrete_unit[
                      input_discrete_unit
                      != self.model.unit_vocab_size - 1
                  ].view(-1, 1)
                  wav = self.vocoder(input_discrete_unit)
                  output_dict.update(wav=wav)
  
      return output_dict
  
imports: import argparse
  import logging
  import shutil
  import sys
  import time
  from pathlib import Path
  from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
  import numpy
  import soundfile
  import torch
  from packaging.version import parse
  from typeguard import check_argument_types
  from espnet2.fileio.datadir_writer import DatadirWriter
  from espnet2.fileio.npy_scp import NpyScpWriter
  from espnet2.tasks.s2st import S2STTask
  from espnet2.text.build_tokenizer import build_tokenizer
  from espnet2.text.token_id_converter import TokenIDConverter
  from espnet2.torch_utils.device_funcs import to_device
  from espnet2.torch_utils.set_all_random_seed import set_all_random_seed
  from espnet2.utils import config_argparse
  from espnet2.utils.types import str2bool, str2triple_str, str_or_none
  from espnet.nets.batch_beam_search import BatchBeamSearch
  from espnet.nets.beam_search import BeamSearch, Hypothesis
  from espnet.nets.pytorch_backend.transformer.subsampling import TooShortUttError
  from espnet.nets.scorer_interface import BatchScorerInterface
  from espnet.nets.scorers.length_bonus import LengthBonus
  from espnet.utils.cli_utils import get_commandline_args
label_desc: Use image node_id 2 for calling the Speech2Speech obj's underlying member method code with example usage: obj.__call__(src_speech, src_speech_lengths, tgt_speech, tgt_speech_lengths, spembs, sids, lids, decode_conf) and returns: output_dict