class_docstr: Speech2Speech class.
docstr: Initialize Speech2Speech module.
code: def __init__(
      self,
      train_config: Union[Path, str] = None,
      model_file: Union[Path, str] = None,
      threshold: float = 0.5,
      minlenratio: float = 0.0,
      maxlenratio: float = 10.0,
      st_subtask_maxlenratio: float = 1.5,
      st_subtask_minlenratio: float = 0.0,
      use_teacher_forcing: bool = False,
      use_att_constraint: bool = False,
      backward_window: int = 1,
      forward_window: int = 3,
      nbest: int = 1,
      normalize_length: bool = False,
      beam_size: int = 5,
      penalty: float = 0.0,
      st_subtask_beam_size: int = 5,
      st_subtask_penalty: float = 0.0,
      st_subtask_nbest: int = 1,
      st_subtask_token_type: str = None,
      st_subtask_bpemodel: str = None,
      vocoder_config: Union[Path, str] = None,
      vocoder_file: Union[Path, str] = None,
      dtype: str = "float32",
      device: str = "cpu",
      seed: int = 777,
      always_fix_seed: bool = False,
      prefer_normalized_feats: bool = False,
  ):
      """Initialize Speech2Speech module."""
      assert check_argument_types()
  
      # setup model
      model, train_args = S2STTask.build_model_from_file(
          train_config, model_file, device
      )
      model.to(dtype=getattr(torch, dtype)).eval()
      self.device = device
      self.dtype = dtype
      self.train_args = train_args
      self.model = model
      self.s2st_type = self.model.s2st_type
      self.preprocess_fn = S2STTask.build_preprocess_fn(
          train_args, False
      )
      self.use_teacher_forcing = use_teacher_forcing
      self.maxlenratio = maxlenratio
      self.minlenratio = minlenratio
      self.st_subtask_maxlenratio = st_subtask_maxlenratio
      self.st_subtask_minlenratio = st_subtask_minlenratio
      self.seed = seed
      self.always_fix_seed = always_fix_seed
      self.prefer_normalized_feats = prefer_normalized_feats
      if self.model.require_vocoder and vocoder_file is not None:
          vocoder = S2STTask.build_vocoder_from_file(
              vocoder_config, vocoder_file, model, device
          )
          if isinstance(vocoder, torch.nn.Module):
              vocoder.to(dtype=getattr(torch, dtype)).eval()
          self.vocoder = vocoder
      else:
          self.vocoder = None
      logging.info(f"S2ST:\n{self.model}")
      if self.vocoder is not None:
          logging.info(f"Vocoder:\n{self.vocoder}")
  
      # setup decoding config
      self.decode_conf = {}  # use for specotrogram-based decoding
      scorers = {}  # use for beam-search decoding
      st_subtask_scorers = {}  # use for beam-search in st_subtask
      if self.s2st_type == "translatotron":
          self.decode_conf.update(
              threshold=threshold,
              maxlenratio=maxlenratio,
              minlenratio=minlenratio,
              use_att_constraint=use_att_constraint,
              forward_window=forward_window,
              backward_window=backward_window,
              use_teacher_forcing=use_teacher_forcing,
          )
      elif (
          self.s2st_type == "discrete_unit" or self.s2st_type == "unity"
      ):
          decoder = model.synthesizer
          token_list = model.unit_token_list
          scorers.update(
              decoder=decoder, length_bonus=LengthBonus(len(token_list))
          )
          weights = dict(
              decoder=1.0,
              length_bonus=penalty,
          )
  
          beam_search = BeamSearch(
              beam_size=beam_size,
              weights=weights,
              scorers=scorers,
              sos=model.unit_sos,
              eos=model.unit_eos,
              vocab_size=len(token_list),
              token_list=None,  # No need to print out the lengthy discrete unit
              pre_beam_score_key="full",
              normalize_length=normalize_length,
          )
  
          # TODO(karita): make all scorers batchfied
          non_batch = [
              k
              for k, v in beam_search.full_scorers.items()
              if not isinstance(v, BatchScorerInterface)
          ]
          if len(non_batch) == 0:
              beam_search.__class__ = BatchBeamSearch
              logging.info(
                  "BatchBeamSearch implementation is selected."
              )
          else:
              logging.warning(
                  f"As non-batch scorers {non_batch} are found, "
                  f"fall back to non-batch implementation."
              )
          beam_search.to(
              device=device, dtype=getattr(torch, dtype)
          ).eval()
          for scorer in scorers.values():
              if isinstance(scorer, torch.nn.Module):
                  scorer.to(
                      device=device, dtype=getattr(torch, dtype)
                  ).eval()
          logging.info(f"Beam_search: {beam_search}")
          logging.info(f"Decoding device={device}, dtype={dtype}")
  
          self.beam_search = beam_search
  
          # Further define st_subtask decoder
          if self.s2st_type == "unity":
              st_subtask_decoder = model.st_decoder
              st_subtask_scorers.update(
                  decoder=st_subtask_decoder,
                  length_bonus=LengthBonus(len(model.tgt_token_list)),
              )
              st_subtask_weights = {
                  "decoder": 1.0,
                  "length_bonus": st_subtask_penalty,
              }
              logging.info("model sos eos: {}".format(model.eos))
              st_subtask_beam_search = BeamSearch(
                  beam_size=st_subtask_beam_size,
                  weights=st_subtask_weights,
                  scorers=st_subtask_scorers,
                  sos=model.sos,
                  eos=model.eos,
                  vocab_size=len(model.tgt_token_list),
                  pre_beam_score_key="full",
                  normalize_length=normalize_length,
                  return_hs=True,
              )
              # TODO(karita): make all st_subtask_scorers batchfied
              non_batch = [
                  k
                  for k, v in st_subtask_beam_search.full_scorers.items()
                  if not isinstance(v, BatchScorerInterface)
              ]
              if len(non_batch) == 0:
                  st_subtask_beam_search.__class__ = BatchBeamSearch
                  logging.info(
                      "BatchBeamSearch implementation is selected."
                  )
              else:
                  logging.warning(
                      f"As non-batch st_subtask_scorers {non_batch} are found, "
                      f"fall back to non-batch implementation."
                  )
              st_subtask_beam_search.to(
                  device=device, dtype=getattr(torch, dtype)
              ).eval()
              for st_subtask_scorers in st_subtask_scorers.values():
                  if isinstance(st_subtask_scorers, torch.nn.Module):
                      st_subtask_scorers.to(
                          device=device, dtype=getattr(torch, dtype)
                      ).eval()
              logging.info(
                  f"st_subtask Beam_search: {st_subtask_beam_search}"
              )
              logging.info(
                  f"st_subtask Decoding device={device}, dtype={dtype}"
              )
  
              self.st_subtask_beam_search = st_subtask_beam_search
  
              # NOTE(jiatong): we here regard the st_subtask as target text
              #                but it may also be source text
              if st_subtask_token_type is None:
                  st_subtask_token_type = train_args.tgt_token_type
              elif st_subtask_token_type == "bpe":
                  if st_subtask_tokenizer is not None:
                      self.st_subtask_tokenizer = build_tokenizer(
                          token_type=st_subtask_token_type,
                          bpemodel=st_subtask_bpemodel,
                      )
                  else:
                      self.st_subtask_tokenizer = None
              else:
                  self.st_subtask_tokenizer = build_tokenizer(
                      token_type=st_subtask_token_type
                  )
              self.st_subtask_converter = TokenIDConverter(
                  token_list=self.model.tgt_token_list
              )
  
      else:
          raise NotImplementedError(
              "Not recognized s2st type of {}".format(self.s2st_type)
          )
  
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 1 to create a new Speech2Speech object  with example: obj = Speech2Speech(train_config, model_file, threshold, minlenratio, maxlenratio, st_subtask_maxlenratio, st_subtask_minlenratio, use_teacher_forcing, use_att_constraint, backward_window, forward_window, nbest, normalize_length, beam_size, penalty, st_subtask_beam_size, st_subtask_penalty, st_subtask_nbest, st_subtask_token_type, st_subtask_bpemodel, vocoder_config, vocoder_file, dtype, device, seed, always_fix_seed, prefer_normalized_feats)