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 apply_frontend(
      self,
      speech: torch.Tensor,
      prev_states=None,
      is_final: bool = False,
  ):
      if prev_states is not None:
          buf = prev_states["waveform_buffer"]
          speech = torch.cat([buf, speech], dim=0)
  
      has_enough_samples = (
          False if speech.size(0) <= self.win_length else True
      )
      if not has_enough_samples:
          if is_final:
              pad = torch.zeros(
                  self.win_length - speech.size(0), dtype=speech.dtype
              )
              speech = torch.cat([speech, pad], dim=0)
          else:
              feats = None
              feats_lengths = None
              next_states = {"waveform_buffer": speech.clone()}
              return feats, feats_lengths, next_states
  
      if is_final:
          speech_to_process = speech
          waveform_buffer = None
      else:
          n_frames = speech.size(0) // self.hop_length
          n_residual = speech.size(0) % self.hop_length
          speech_to_process = speech.narrow(
              0, 0, n_frames * self.hop_length
          )
          waveform_buffer = speech.narrow(
              0,
              speech.size(0)
              - (
                  math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  )
                  * 2
                  - 1
              )
              * self.hop_length
              - n_residual,
              (
                  math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  )
                  * 2
                  - 1
              )
              * self.hop_length
              + n_residual,
          ).clone()
  
      # data: (Nsamples,) -> (1, Nsamples)
      speech_to_process = speech_to_process.unsqueeze(0).to(
          getattr(torch, self.dtype)
      )
      lengths = speech_to_process.new_full(
          [1], dtype=torch.long, fill_value=speech_to_process.size(1)
      )
      batch = {"speech": speech_to_process, "speech_lengths": lengths}
  
      # lenghts: (1,)
      # a. To device
      batch = to_device(batch, device=self.device)
  
      feats, feats_lengths = self.asr_model._extract_feats(**batch)
      if self.asr_model.normalize is not None:
          feats, feats_lengths = self.asr_model.normalize(
              feats, feats_lengths
          )
  
      # Trimming
      if is_final:
          if prev_states is None:
              pass
          else:
              feats = feats.narrow(
                  1,
                  math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  ),
                  feats.size(1)
                  - math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  ),
              )
      else:
          if prev_states is None:
              feats = feats.narrow(
                  1,
                  0,
                  feats.size(1)
                  - math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  ),
              )
          else:
              feats = feats.narrow(
                  1,
                  math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  ),
                  feats.size(1)
                  - 2
                  * math.ceil(
                      math.ceil(self.win_length / self.hop_length) / 2
                  ),
              )
  
      feats_lengths = feats.new_full(
          [1], dtype=torch.long, fill_value=feats.size(1)
      )
  
      if is_final:
          next_states = None
      else:
          next_states = {"waveform_buffer": waveform_buffer}
      return feats, feats_lengths, next_states
  
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 3 for calling the Speech2TextStreaming obj's underlying member method code with example usage: obj.apply_frontend(speech, prev_states, is_final) and returns: feats,  feats_lengths,  next_states, feats,  feats_lengths,  next_states