class_docstr: Online beam search implementation.
  
  This simulates streaming decoding.
  It requires encoded features of entire utterance and
  extracts block by block from it as it shoud be done
  in streaming processing.
  This is based on Tsunoo et al, "STREAMING TRANSFORMER ASR
  WITH BLOCKWISE SYNCHRONOUS BEAM SEARCH"
  (https://arxiv.org/abs/2006.14941).
docstr: Perform beam search.
  
  Args:
      x (torch.Tensor): Encoded speech feature (T, D)
      maxlenratio (float): Input length ratio to obtain max output length.
          If maxlenratio=0.0 (default), it uses a end-detect function
          to automatically find maximum hypothesis lengths
      minlenratio (float): Input length ratio to obtain min output length.
  
  Returns:
      list[Hypothesis]: N-best decoding results
docstr_tok: ["Perform","beam","search",".","Args",":","x","(","torch.Tensor",")",":","Encoded","speech","feature","(","T",",","D",")","maxlenratio","(","float",")",":","Input","length","ratio","to","obtain","max","output","length",".","If","maxlenratio=0.0","(","default",")",",","it","uses","a","end-detect","function","to","automatically","find","maximum","hypothesis","lengths","minlenratio","(","float",")",":","Input","length","ratio","to","obtain","min","output","length",".","Returns",":","list","[","Hypothesis","]",":","N-best","decoding","results"]
code: def forward(
      self,
      x: torch.Tensor,
      maxlenratio: float = 0.0,
      minlenratio: float = 0.0,
      is_final: bool = True,
  ) -> List[Hypothesis]:
      """Perform beam search.
  
      Args:
          x (torch.Tensor): Encoded speech feature (T, D)
          maxlenratio (float): Input length ratio to obtain max output length.
              If maxlenratio=0.0 (default), it uses a end-detect function
              to automatically find maximum hypothesis lengths
          minlenratio (float): Input length ratio to obtain min output length.
  
      Returns:
          list[Hypothesis]: N-best decoding results
  
      """
      if self.encbuffer is None or self.block_size == 0:
          self.encbuffer = x
      else:
          self.encbuffer = torch.cat([self.encbuffer, x], axis=0)
  
      x = self.encbuffer
  
      # set length bounds
      if maxlenratio == 0:
          maxlen = x.shape[0]
      else:
          maxlen = max(1, int(maxlenratio * x.size(0)))
  
      if minlenratio < 0:
          minlen = -1 * int(minlenratio)
      else:
          minlen = int(minlenratio * x.size(0))
  
      # set block_size == 0 for recomputing
      if self.block_size == 0:
          block_is_final = is_final
          h = x
          logging.debug(
              "  Feature length: {}, current position: {}".format(
                  h.shape[0], self.process_idx
              )
          )
  
          if self.running_hyps is None:  # init hyps
              init_scores = {}
              init_states = {}
              if "ctc" in self.scorers.keys():
                  self.scorers["ctc"].batch_init_state(h)
                  init_scores["ctc"] = torch.tensor([0.0])
                  init_states["ctc"] = [None]
              if "length_bonus" in self.scorers.keys():
                  init_scores["length_bonus"] = torch.tensor([0.0])
                  init_states["length_bonus"] = [None]
              if "decoder" in self.scorers.keys():
                  init_scores["decoder"] = torch.tensor([0.0])
                  init_states["decoder"] = [None]
              self.running_hyps = BatchHypothesis(
                  score=torch.tensor([0.0]),
                  scores=init_scores,
                  states=init_states,
                  length=torch.tensor([2]),
                  yseq=torch.tensor([self.hyp_primer], device=x.device),
                  hs=[],
              )
              self.prev_incremental = self.running_hyps
  
          if self.time_sync:
              ret = self.process_one_block_time_sync(
                  h, block_is_final, maxlen, maxlenratio
              )
          else:
              ret = self.process_one_block(
                  h,
                  block_is_final,
                  maxlen - self.process_idx,
                  maxlenratio,
              )
          logging.debug(
              "Finished processing chunk: %d", self.processed_block
          )
          self.processed_block += 1
  
          # prune running_hyps, taking top as an incremental decoding
          if self.incremental_decode and not is_final:
              if (
                  self.running_hyps.yseq.shape[0] == 0
              ):  # running_hyps will be empty if maxlen is reached
                  logging.info(
                      "search stopped by maxlen in a non final chunk. \
                          reverting to prev running hyp"
                  )
                  self.running_hyps = self.prev_incremental
              logging.info(
                  "Hyps before incremental pruning: %d",
                  self.running_hyps.yseq.shape[0],
              )
              if self.running_hyps.yseq.shape[0] > 0:
                  self.running_hyps = self._batch_select(
                      self.running_hyps, [0]
                  )
                  self.prev_incremental = self.running_hyps
              logging.info(
                  "Hyps after incremental pruning: %d",
                  self.running_hyps.yseq.shape[0],
              )
  
              if self.token_list is not None:
                  logging.info(
                      "best running hypo: "
                      + "".join(
                          [
                              self.token_list[x]
                              for x in self.running_hyps.yseq[0, 1:]
                          ]
                      )
                  )
  
              # hold_n
              if self.hold_n > 0 and self.running_hyps.length[0] > 2:
                  self.running_hyps = BatchHypothesis(
                      score=self.running_hyps.score,
                      scores=self.running_hyps.scores,
                      states=self.running_hyps.states,
                      length=self.running_hyps.length - self.hold_n,
                      yseq=self.running_hyps.yseq[:, : -self.hold_n],
                      hs=[],
                  )
                  if self.token_list is not None:
                      logging.info(
                          "best hypo after hold: "
                          + "".join(
                              [
                                  self.token_list[x]
                                  for x in self.running_hyps.yseq[0, 1:]
                              ]
                          )
                      )
  
          if is_final:
              if len(ret) == 0:
                  if self.prev_output is None:
                      return []
                  else:
                      return self.prev_output
              else:
                  return ret
          else:
              # dont return incremental hyps,
              # check them by grabbing top running_hyp
              if len(ret) > 0:
                  self.prev_output = ret
              return []
  
      # blockwise processing w/ rewinding
      else:
          ret = None
          while True:
              cur_end_frame = (
                  self.block_size
                  - self.look_ahead
                  + self.hop_size * self.processed_block
              )
              if cur_end_frame < x.shape[0]:
                  h = x.narrow(0, 0, cur_end_frame)
                  block_is_final = False
              else:
                  if is_final:
                      h = x
                      block_is_final = True
                  else:
                      break
  
              logging.debug(
                  "Start processing block: %d", self.processed_block
              )
              logging.debug(
                  "  Feature length: {}, current position: {}".format(
                      h.shape[0], self.process_idx
                  )
              )
              if (
                  self.encoded_feat_length_limit > 0
                  and h.shape[0] > self.encoded_feat_length_limit
              ):
                  h = h.narrow(
                      0,
                      h.shape[0] - self.encoded_feat_length_limit,
                      self.encoded_feat_length_limit,
                  )
  
              if self.running_hyps is None:
                  self.running_hyps = self.init_hyp(h)
              if self.time_sync:
                  ret = self.process_one_block_time_sync(
                      h, block_is_final, maxlen, maxlenratio
                  )
              else:
                  ret = self.process_one_block(
                      h, block_is_final, maxlen, minlen, maxlenratio
                  )
              logging.debug(
                  "Finished processing block: %d", self.processed_block
              )
              self.processed_block += 1
  
              # prune running_hyps, taking top as an incremental decoding
              if self.incremental_decode:
                  logging.debug(
                      "Hyps before incremental pruning: %d",
                      self.running_hyps.yseq.shape[0],
                  )
                  if self.running_hyps.yseq.shape[0] > 0:
                      self.running_hyps = self._batch_select(
                          self.running_hyps, [0]
                      )
                  logging.debug(
                      "Hyps after incremental pruning: %d",
                      self.running_hyps.yseq.shape[0],
                  )
  
              if block_is_final:
                  return ret
          if ret is None:
              if self.prev_output is None:
                  return []
              else:
                  return self.prev_output
          else:
              self.prev_output = ret
              # N-best results
              return ret
  
imports: import logging
  from typing import Any
  from typing import Dict
  from typing import List
  from typing import Tuple
  import torch
  from espnet2.asr.transducer.beam_search_transducer_streaming import BeamSearchTransducerStreaming
  from espnet.nets.batch_beam_search import BatchBeamSearch
  from espnet.nets.batch_beam_search import BatchHypothesis
  from espnet.nets.beam_search import Hypothesis
  from espnet.nets.beam_search_timesync_streaming import BeamSearchTimeSyncStreaming
  from espnet.nets.e2e_asr_common import end_detect
label_desc: Use image node_id 4 for calling the BatchBeamSearchOnline obj's underlying member method code with example usage: obj.forward(x, maxlenratio, minlenratio, is_final) and returns: list, ret, ret, ret, list, self, list, self