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), ...]
class_docstr_tok: ["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,
      asr_train_config: Union[Path, str],
      asr_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.5,
      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,
  ):
      assert check_argument_types()
  
      # 1. Build ASR model
      scorers = {}
      asr_model, asr_train_args = ASRTask.build_model_from_file(
          asr_train_config, asr_model_file, device
      )
      asr_model.to(dtype=getattr(torch, dtype)).eval()
  
      assert isinstance(
          asr_model.encoder, ContextualBlockTransformerEncoder
      ) or isinstance(
          asr_model.encoder, ContextualBlockConformerEncoder
      )
  
      decoder = asr_model.decoder
      ctc = CTCPrefixScorer(ctc=asr_model.ctc, eos=asr_model.eos)
      token_list = asr_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,
      )
  
      assert "encoder_conf" in asr_train_args
      assert "look_ahead" in asr_train_args.encoder_conf
      assert "hop_size" in asr_train_args.encoder_conf
      assert "block_size" in asr_train_args.encoder_conf
      # look_ahead = asr_train_args.encoder_conf['look_ahead']
      # hop_size   = asr_train_args.encoder_conf['hop_size']
      # block_size = asr_train_args.encoder_conf['block_size']
  
      assert batch_size == 1
  
      beam_search = BatchBeamSearchOnline(
          beam_size=beam_size,
          weights=weights,
          scorers=scorers,
          sos=asr_model.sos,
          eos=asr_model.eos,
          vocab_size=len(token_list),
          token_list=token_list,
          pre_beam_score_key=None if ctc_weight == 1.0 else "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,
      )
  
      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 = asr_train_args.token_type
      if bpemodel is None:
          bpemodel = asr_train_args.bpemodel
  
      if token_type is None:
          tokenizer = None
      elif token_type == "bpe":
          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.asr_model = asr_model
      self.asr_train_args = asr_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 asr_train_args.frontend_conf:
          self.n_fft = asr_train_args.frontend_conf["n_fft"]
      else:
          self.n_fft = 512
      if "hop_length" in asr_train_args.frontend_conf:
          self.hop_length = asr_train_args.frontend_conf["hop_length"]
      else:
          self.hop_length = 128
      if (
          "win_length" in asr_train_args.frontend_conf
          and asr_train_args.frontend_conf["win_length"] is not None
      ):
          self.win_length = asr_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.fileio.datadir_writer import DatadirWriter
  from espnet2.tasks.asr import ASRTask
  from espnet2.tasks.lm import LMTask
  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(asr_train_config, asr_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)