class_docstr: Beam search implementation for Transducer.
docstr: It's the modified Adaptive Expansion Search (mAES) implementation.
  
  Based on/modified from https://ieeexplore.ieee.org/document/9250505 and NSC.
  
  Args:
      enc_out: Encoder output sequence. (T, D_enc)
  
  Returns:
      nbest_hyps: N-best hypothesis.
docstr_tok: ["It","'s","the","modified","Adaptive","Expansion","Search","(","mAES",")","implementation",".","Based","on\/modified","from","https",":","\/\/ieeexplore.ieee.org\/document\/9250505","and","NSC",".","Args",":","enc_out",":","Encoder","output","sequence",".","(","T",",","D_enc",")","Returns",":","nbest_hyps",":","N-best","hypothesis","."]
code: def modified_adaptive_expansion_search(
      self, enc_out: torch.Tensor
  ) -> List[ExtendedHypothesis]:
      """It's the modified Adaptive Expansion Search (mAES) implementation.
  
      Based on/modified from https://ieeexplore.ieee.org/document/9250505 and NSC.
  
      Args:
          enc_out: Encoder output sequence. (T, D_enc)
  
      Returns:
          nbest_hyps: N-best hypothesis.
  
      """
      beam = min(self.beam_size, self.vocab_size)
      beam_state = self.decoder.init_state(beam)
  
      init_tokens = [
          ExtendedHypothesis(
              yseq=[self.blank_id],
              score=0.0,
              dec_state=self.decoder.select_state(beam_state, 0),
          )
      ]
  
      cache = {}
  
      (
          beam_dec_out,
          beam_state,
          beam_lm_tokens,
      ) = self.decoder.batch_score(
          init_tokens,
          beam_state,
          cache,
          self.use_lm,
      )
  
      state = self.decoder.select_state(beam_state, 0)
  
      if self.use_lm:
          beam_lm_scores, beam_lm_states = self.lm.batch_score(
              beam_lm_tokens, [i.lm_state for i in init_tokens], None
          )
  
          lm_state = beam_lm_states[0]
          lm_scores = beam_lm_scores[0]
      else:
          lm_state = None
          lm_scores = None
  
      kept_hyps = [
          ExtendedHypothesis(
              yseq=[self.blank_id],
              score=0.0,
              dec_state=state,
              dec_out=[beam_dec_out[0]],
              lm_state=lm_state,
              lm_scores=lm_scores,
          )
      ]
  
      for enc_out_t in enc_out:
          hyps = self.prefix_search(
              sorted(
                  kept_hyps, key=lambda x: len(x.yseq), reverse=True
              ),
              enc_out_t,
          )
          kept_hyps = []
  
          beam_enc_out = enc_out_t.unsqueeze(0)
  
          list_b = []
          duplication_check = [hyp.yseq for hyp in hyps]
  
          for n in range(self.nstep):
              beam_dec_out = torch.stack([h.dec_out[-1] for h in hyps])
  
              beam_logp, beam_idx = torch.log_softmax(
                  self.joint_network(beam_enc_out, beam_dec_out),
                  dim=-1,
              ).topk(self.max_candidates, dim=-1)
  
              k_expansions = select_k_expansions(
                  hyps,
                  beam_idx,
                  beam_logp,
                  self.expansion_gamma,
              )
  
              list_exp = []
              for i, hyp in enumerate(hyps):
                  for k, new_score in k_expansions[i]:
                      new_hyp = ExtendedHypothesis(
                          yseq=hyp.yseq[:],
                          score=new_score,
                          dec_out=hyp.dec_out[:],
                          dec_state=hyp.dec_state,
                          lm_state=hyp.lm_state,
                          lm_scores=hyp.lm_scores,
                      )
  
                      if k == 0:
                          list_b.append(new_hyp)
                      else:
                          if (
                              new_hyp.yseq + [int(k)]
                              not in duplication_check
                          ):
                              new_hyp.yseq.append(int(k))
  
                              if self.use_lm:
                                  new_hyp.score += (
                                      self.lm_weight
                                      * float(hyp.lm_scores[k])
                                  )
  
                              list_exp.append(new_hyp)
  
              if not list_exp:
                  kept_hyps = sorted(
                      list_b, key=lambda x: x.score, reverse=True
                  )[:beam]
  
                  break
              else:
                  beam_state = self.decoder.create_batch_states(
                      beam_state,
                      [hyp.dec_state for hyp in list_exp],
                      [hyp.yseq for hyp in list_exp],
                  )
  
                  (
                      beam_dec_out,
                      beam_state,
                      beam_lm_tokens,
                  ) = self.decoder.batch_score(
                      list_exp,
                      beam_state,
                      cache,
                      self.use_lm,
                  )
  
                  if self.use_lm:
                      (
                          beam_lm_scores,
                          beam_lm_states,
                      ) = self.lm.batch_score(
                          beam_lm_tokens,
                          [k.lm_state for k in list_exp],
                          None,
                      )
  
                  if n < (self.nstep - 1):
                      for i, hyp in enumerate(list_exp):
                          hyp.dec_out.append(beam_dec_out[i])
                          hyp.dec_state = self.decoder.select_state(
                              beam_state, i
                          )
  
                          if self.use_lm:
                              hyp.lm_state = beam_lm_states[i]
                              hyp.lm_scores = beam_lm_scores[i]
  
                      hyps = list_exp[:]
                  else:
                      beam_logp = torch.log_softmax(
                          self.joint_network(
                              beam_enc_out, beam_dec_out
                          ),
                          dim=-1,
                      )
  
                      for i, hyp in enumerate(list_exp):
                          hyp.score += float(beam_logp[i, 0])
  
                          hyp.dec_out.append(beam_dec_out[i])
                          hyp.dec_state = self.decoder.select_state(
                              beam_state, i
                          )
  
                          if self.use_lm:
                              hyp.lm_states = beam_lm_states[i]
                              hyp.lm_scores = beam_lm_scores[i]
  
                      kept_hyps = sorted(
                          list_b + list_exp,
                          key=lambda x: x.score,
                          reverse=True,
                      )[:beam]
  
      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 11 for calling the BeamSearchTransducerStreaming obj's underlying member method code with example usage: obj.modified_adaptive_expansion_search(enc_out) and returns: self