class_docstr: Speech2TextStreaming class
  
  Details in "Streaming Transformer ASR with Blockwise Synchronous Beam Search"
  (https://arxiv.org/abs/2006.14941)
  
  Examples:
      >>> import soundfile
      >>> speech2text = Speech2TextStreaming("asr_config.yml", "asr.pth")
      >>> audio, rate = soundfile.read("speech.wav")
      >>> speech2text(audio)
      [(text, token, token_int, hypothesis object), ...]
code: def __init__(
      self,
      st_train_config: Union[Path, str],
      st_model_file: Union[Path, str] = None,
      lm_train_config: Union[Path, str] = None,
      lm_file: Union[Path, str] = None,
      token_type: str = None,
      bpemodel: str = None,
      device: str = "cpu",
      maxlenratio: float = 0.0,
      minlenratio: float = 0.0,
      batch_size: int = 1,
      dtype: str = "float32",
      beam_size: int = 20,
      ctc_weight: float = 0.0,
      lm_weight: float = 1.0,
      penalty: float = 0.0,
      nbest: int = 1,
      normalize_length: bool = False,
      disable_repetition_detection=False,
      decoder_text_length_limit=0,
      encoded_feat_length_limit=0,
      time_sync: bool = False,
      incremental_decode: bool = False,
      blank_penalty: float = 1.0,
      hold_n: int = 0,
      transducer_conf: dict = None,
      hugging_face_decoder: bool = False,
  ):
      assert check_argument_types()
  
      # 1. Build ST model
      scorers = {}
      st_model, st_train_args = STTask.build_model_from_file(
          st_train_config, st_model_file, device
      )
      st_model.to(dtype=getattr(torch, dtype)).eval()
  
      if isinstance(
          st_model.encoder, ContextualBlockTransformerEncoder
      ) or isinstance(
          st_model.encoder, ContextualBlockConformerEncoder
      ):
          if isinstance(st_model.frontend, S3prlFrontend):
              raise NotImplementedError(
                  "S3prlFrontend not supported with blockwise encoder"
              )
          if st_model.hier_encoder is not None:
              raise NotImplementedError(
                  "hierarchical encoder not supported with blockwise encoder"
              )
          block_size = st_train_args.encoder_conf["block_size"]
      else:
          block_size = 0  # recompute encoder with every new chunk
  
      decoder = st_model.decoder
      if hasattr(st_model, "st_ctc"):
          ctc = CTCPrefixScorer(ctc=st_model.st_ctc, eos=st_model.eos)
      else:
          ctc = None
  
      token_list = st_model.token_list
      scorers.update(
          decoder=decoder,
          ctc=ctc,
          length_bonus=LengthBonus(len(token_list)),
      )
  
      # 2. Build Language model
      if lm_train_config is not None:
          lm, lm_train_args = LMTask.build_model_from_file(
              lm_train_config, lm_file, device
          )
          scorers["lm"] = lm.lm
  
      # 3. Build BeamSearch object
      weights = dict(
          decoder=1.0 - ctc_weight,
          ctc=ctc_weight,
          lm=lm_weight,
          length_bonus=penalty,
          blank_penalty=blank_penalty,
      )
  
      # assert "encoder_conf" in st_train_args
      # assert "look_ahead" in st_train_args.encoder_conf
      # assert "hop_size" in st_train_args.encoder_conf
      # assert "block_size" in st_train_args.encoder_conf
  
      assert batch_size == 1
  
      if (
          decoder.__class__.__name__ == "HuggingFaceTransformersDecoder"
          and hugging_face_decoder
      ):
          if not is_transformers_available:
              raise ImportError(
                  "`transformers` is not available."
                  " Please install it via `pip install transformers`"
                  " or `cd /path/to/espnet/tools && . ./activate_python.sh"
                  " && ./installers/install_transformers.sh`."
              )
  
          hugging_face_model = AutoModelForSeq2SeqLM.from_pretrained(
              decoder.model_name_or_path
          )
  
          hugging_face_model.lm_head.load_state_dict(
              decoder.lm_head.state_dict()
          )
  
          if hasattr(hugging_face_model, "model"):
              hugging_face_model.model.decoder.load_state_dict(
                  decoder.decoder.state_dict()
              )
              del hugging_face_model.model.encoder
          else:
              hugging_face_model.decoder.load_state_dict(
                  decoder.decoder.state_dict()
              )
              del hugging_face_model.encoder
  
          # del st_model.decoder.lm_head
          # del st_model.decoder.decoder
  
          hugging_face_linear_in = decoder.linear_in
          hugging_face_model.to(device=device).eval()
  
          # hacky way to use .score()
          st_model.decoder.hf_generate = hugging_face_model
          weights = dict(
              decoder=1.0 - ctc_weight,
              ctc=ctc_weight,
              lm=lm_weight,
              length_bonus=penalty,
          )
          beam_search = BatchBeamSearchOnline(
              beam_size=beam_size,
              weights=weights,
              scorers=scorers,
              sos=hugging_face_model.config.decoder_start_token_id,
              eos=hugging_face_model.config.eos_token_id,
              vocab_size=len(token_list),
              token_list=token_list,
              pre_beam_score_key="full",
              normalize_length=normalize_length,
              disable_repetition_detection=disable_repetition_detection,
              decoder_text_length_limit=decoder_text_length_limit,
              encoded_feat_length_limit=encoded_feat_length_limit,
              incremental_decode=incremental_decode,
              time_sync=time_sync,
              block_size=block_size,
              ctc=st_model.st_ctc
              if hasattr(st_model, "st_ctc")
              else None,
              hold_n=hold_n,
          )
          self.hugging_face_model = hugging_face_model
          self.hugging_face_linear_in = hugging_face_linear_in
  
      else:
          beam_search = BatchBeamSearchOnline(
              beam_size=beam_size,
              weights=weights,
              scorers=scorers,
              sos=st_model.sos,
              eos=st_model.eos,
              vocab_size=len(token_list),
              token_list=token_list,
              pre_beam_score_key="full",
              disable_repetition_detection=disable_repetition_detection,
              decoder_text_length_limit=decoder_text_length_limit,
              encoded_feat_length_limit=encoded_feat_length_limit,
              incremental_decode=incremental_decode,
              time_sync=time_sync,
              ctc=st_model.st_ctc
              if hasattr(st_model, "st_ctc")
              else None,
              hold_n=hold_n,
              transducer_conf=transducer_conf,
              joint_network=st_model.st_joint_network
              if hasattr(st_model, "st_joint_network")
              else None,
          )
          self.hugging_face_model = None
          self.hugging_face_linear_in = None
  
      if transducer_conf is None:
          non_batch = [
              k
              for k, v in beam_search.full_scorers.items()
              if not isinstance(v, BatchScorerInterface)
          ]
          assert len(non_batch) == 0
  
          # TODO(karita): make all scorers batchfied
          logging.info(
              "BatchBeamSearchOnline implementation is selected."
          )
  
      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}")
  
      # 4. [Optional] Build Text converter: e.g. bpe-sym -> Text
      if token_type is None:
          token_type = st_train_args.token_type
      if bpemodel is None:
          bpemodel = st_train_args.bpemodel
  
      if token_type is None:
          tokenizer = None
      elif token_type == "bpe" or token_type == "hugging_face":
          if bpemodel is not None:
              tokenizer = build_tokenizer(
                  token_type=token_type, bpemodel=bpemodel
              )
          else:
              tokenizer = None
      else:
          tokenizer = build_tokenizer(token_type=token_type)
      converter = TokenIDConverter(token_list=token_list)
      logging.info(f"Text tokenizer: {tokenizer}")
  
      self.st_model = st_model
      self.st_train_args = st_train_args
      self.converter = converter
      self.tokenizer = tokenizer
      self.beam_search = beam_search
      self.maxlenratio = maxlenratio
      self.minlenratio = minlenratio
      self.device = device
      self.dtype = dtype
      self.nbest = nbest
      if "n_fft" in st_train_args.frontend_conf:
          self.n_fft = st_train_args.frontend_conf["n_fft"]
      else:
          self.n_fft = 512
      if "hop_length" in st_train_args.frontend_conf:
          self.hop_length = st_train_args.frontend_conf["hop_length"]
      else:
          self.hop_length = 128
      if (
          "win_length" in st_train_args.frontend_conf
          and st_train_args.frontend_conf["win_length"] is not None
      ):
          self.win_length = st_train_args.frontend_conf["win_length"]
      else:
          self.win_length = self.n_fft
  
      self.reset()
  
imports: import argparse
  import logging
  import math
  import sys
  from pathlib import Path
  from typing import List, Optional, Sequence, Tuple, Union
  import numpy
  import torch
  from typeguard import check_argument_types, check_return_type
  from espnet2.asr.encoder.contextual_block_conformer_encoder import ContextualBlockConformerEncoder
  from espnet2.asr.encoder.contextual_block_transformer_encoder import ContextualBlockTransformerEncoder
  from espnet2.asr.frontend.s3prl import S3prlFrontend
  from espnet2.fileio.datadir_writer import DatadirWriter
  from espnet2.tasks.lm import LMTask
  from espnet2.tasks.st import STTask
  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_online import BatchBeamSearchOnline
  from espnet.nets.beam_search import Hypothesis
  from espnet.nets.pytorch_backend.transformer.subsampling import TooShortUttError
  from espnet.nets.scorer_interface import BatchScorerInterface
  from espnet.nets.scorers.ctc import CTCPrefixScorer
  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 Speech2TextStreaming object  with example: obj = Speech2TextStreaming(st_train_config, st_model_file, lm_train_config, lm_file, token_type, bpemodel, device, maxlenratio, minlenratio, batch_size, dtype, beam_size, ctc_weight, lm_weight, penalty, nbest, normalize_length, disable_repetition_detection, decoder_text_length_limit, encoded_feat_length_limit, time_sync, incremental_decode, blank_penalty, hold_n, transducer_conf, hugging_face_decoder)