class_docstr: Beam search implementation for Transducer.
docstr: Beam search implementation.
  
  Modified from https://arxiv.org/pdf/1211.3711.pdf
  
  Args:
      enc_out: Encoder output sequence. (T, D)
  
  Returns:
      nbest_hyps: N-best hypothesis.
docstr_tok: ["Beam","search","implementation",".","Modified","from","https",":","\/\/arxiv.org\/pdf\/1211.3711.pdf","Args",":","enc_out",":","Encoder","output","sequence",".","(","T",",","D",")","Returns",":","nbest_hyps",":","N-best","hypothesis","."]
code: def default_beam_search(
      self, enc_out: torch.Tensor
  ) -> List[Hypothesis]:
      """Beam search implementation.
  
      Modified from https://arxiv.org/pdf/1211.3711.pdf
  
      Args:
          enc_out: Encoder output sequence. (T, D)
  
      Returns:
          nbest_hyps: N-best hypothesis.
  
      """
      beam = min(self.beam_size, self.vocab_size)
      beam_k = min(beam, (self.vocab_size - 1))
  
      dec_state = self.decoder.init_state(1)
  
      kept_hyps = [
          Hypothesis(
              score=0.0, yseq=[self.blank_id], dec_state=dec_state
          )
      ]
      cache = {}
      cache_lm = {}
  
      for enc_out_t in enc_out:
          hyps = kept_hyps
          kept_hyps = []
  
          if self.token_list is not None:
              logging.debug(
                  "\n"
                  + "\n".join(
                      [
                          "hypo: "
                          + "".join(
                              [self.token_list[x] for x in hyp.yseq[1:]]
                          )
                          + f", score: {round(float(hyp.score), 2)}"
                          for hyp in sorted(
                              hyps, key=lambda x: x.score, reverse=True
                          )
                      ]
                  )
              )
  
          while True:
              if self.score_norm_during:
                  max_hyp = max(
                      hyps, key=lambda x: x.score / len(x.yseq)
                  )
              else:
                  max_hyp = max(hyps, key=lambda x: x.score)
              hyps.remove(max_hyp)
  
              dec_out, state, lm_tokens = self.decoder.score(
                  max_hyp, cache
              )
  
              logp = torch.log_softmax(
                  self.joint_network(enc_out_t, dec_out),
                  dim=-1,
              )
              top_k = logp[1:].topk(beam_k, dim=-1)
  
              kept_hyps.append(
                  Hypothesis(
                      score=(max_hyp.score + float(logp[0:1])),
                      yseq=max_hyp.yseq[:],
                      dec_state=max_hyp.dec_state,
                      lm_state=max_hyp.lm_state,
                  )
              )
  
              if self.use_lm:
                  if tuple(max_hyp.yseq) not in cache_lm:
                      lm_scores, lm_state = self.lm.score(
                          torch.LongTensor(
                              [self.sos] + max_hyp.yseq[1:],
                              device=self.decoder.device,
                          ),
                          max_hyp.lm_state,
                          None,
                      )
                      cache_lm[tuple(max_hyp.yseq)] = (
                          lm_scores,
                          lm_state,
                      )
                  else:
                      lm_scores, lm_state = cache_lm[
                          tuple(max_hyp.yseq)
                      ]
              else:
                  lm_state = max_hyp.lm_state
  
              for logp, k in zip(*top_k):
                  score = max_hyp.score + float(logp)
  
                  if self.use_lm:
                      score += self.lm_weight * lm_scores[k + 1]
  
                  hyps.append(
                      Hypothesis(
                          score=score,
                          yseq=max_hyp.yseq[:] + [int(k + 1)],
                          dec_state=state,
                          lm_state=lm_state,
                      )
                  )
  
              if self.score_norm_during:
                  hyps_max = float(
                      max(
                          hyps, key=lambda x: x.score / len(x.yseq)
                      ).score
                  )
              else:
                  hyps_max = float(
                      max(hyps, key=lambda x: x.score).score
                  )
              kept_most_prob = sorted(
                  [hyp for hyp in kept_hyps if hyp.score > hyps_max],
                  key=lambda x: x.score,
              )
              if len(kept_most_prob) >= beam:
                  kept_hyps = kept_most_prob
                  break
  
      return self.sort_nbest(kept_hyps)
  
imports: import logging
  from dataclasses import dataclass
  from typing import Any, Dict, List, Optional, Tuple, Union
  import numpy
  import torch
  from espnet2.asr.decoder.abs_decoder import AbsDecoder
  from espnet2.asr_transducer.joint_network import JointNetwork
  from espnet2.lm.transformer_lm import TransformerLM
  from espnet.nets.beam_search import Hypothesis
  from espnet.nets.pytorch_backend.transducer.utils import is_prefix, recombine_hyps, select_k_expansions, subtract
label_desc: Use image node_id 7 for calling the BeamSearchTransducerStreaming obj's underlying member method code with example usage: obj.default_beam_search(enc_out) and returns: self