class_docstr: Beam search implementation for Transducer.
docstr: N-step constrained beam search implementation.
  
  Based on/Modified from https://arxiv.org/pdf/2002.03577.pdf.
  Please reference ESPnet (b-flo, PR #2444) for any usage outside ESPnet
  until further modifications.
  
  Args:
      enc_out: Encoder output sequence. (T, D_enc)
  
  Returns:
      nbest_hyps: N-best hypothesis.
docstr_tok: ["N-step","constrained","beam","search","implementation",".","Based","on\/Modified","from","https",":","\/\/arxiv.org\/pdf\/2002.03577.pdf",".","Please","reference","ESPnet","(","b-flo",",","PR","#","2444",")","for","any","usage","outside","ESPnet","until","further","modifications",".","Args",":","enc_out",":","Encoder","output","sequence",".","(","T",",","D_enc",")","Returns",":","nbest_hyps",":","N-best","hypothesis","."]
code: def nsc_beam_search(
      self, enc_out: torch.Tensor
  ) -> List[ExtendedHypothesis]:
      """N-step constrained beam search implementation.
  
      Based on/Modified from https://arxiv.org/pdf/2002.03577.pdf.
      Please reference ESPnet (b-flo, PR #2444) for any usage outside ESPnet
      until further modifications.
  
      Args:
          enc_out: Encoder output sequence. (T, D_enc)
  
      Returns:
          nbest_hyps: N-best hypothesis.
  
      """
      beam = min(self.beam_size, self.vocab_size)
      beam_k = min(beam, (self.vocab_size - 1))
  
      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)
  
          S = []
          V = []
          for n in range(self.nstep):
              beam_dec_out = torch.stack(
                  [hyp.dec_out[-1] for hyp in hyps]
              )
  
              beam_logp = torch.log_softmax(
                  self.joint_network(beam_enc_out, beam_dec_out),
                  dim=-1,
              )
              beam_topk = beam_logp[:, 1:].topk(beam_k, dim=-1)
  
              for i, hyp in enumerate(hyps):
                  S.append(
                      ExtendedHypothesis(
                          yseq=hyp.yseq[:],
                          score=hyp.score + float(beam_logp[i, 0:1]),
                          dec_out=hyp.dec_out[:],
                          dec_state=hyp.dec_state,
                          lm_state=hyp.lm_state,
                          lm_scores=hyp.lm_scores,
                      )
                  )
  
                  for logp, k in zip(
                      beam_topk[0][i], beam_topk[1][i] + 1
                  ):
                      score = hyp.score + float(logp)
  
                      if self.use_lm:
                          score += self.lm_weight * float(
                              hyp.lm_scores[k]
                          )
  
                      V.append(
                          ExtendedHypothesis(
                              yseq=hyp.yseq[:] + [int(k)],
                              score=score,
                              dec_out=hyp.dec_out[:],
                              dec_state=hyp.dec_state,
                              lm_state=hyp.lm_state,
                              lm_scores=hyp.lm_scores,
                          )
                      )
  
              V.sort(key=lambda x: x.score, reverse=True)
              V = subtract(V, hyps)[:beam]
  
              beam_state = self.decoder.create_batch_states(
                  beam_state,
                  [v.dec_state for v in V],
                  [v.yseq for v in V],
              )
              (
                  beam_dec_out,
                  beam_state,
                  beam_lm_tokens,
              ) = self.decoder.batch_score(
                  V,
                  beam_state,
                  cache,
                  self.use_lm,
              )
  
              if self.use_lm:
                  beam_lm_scores, beam_lm_states = self.lm.batch_score(
                      beam_lm_tokens, [v.lm_state for v in V], None
                  )
  
              if n < (self.nstep - 1):
                  for i, v in enumerate(V):
                      v.dec_out.append(beam_dec_out[i])
  
                      v.dec_state = self.decoder.select_state(
                          beam_state, i
                      )
  
                      if self.use_lm:
                          v.lm_state = beam_lm_states[i]
                          v.lm_scores = beam_lm_scores[i]
  
                  hyps = V[:]
              else:
                  beam_logp = torch.log_softmax(
                      self.joint_network(beam_enc_out, beam_dec_out),
                      dim=-1,
                  )
  
                  for i, v in enumerate(V):
                      if self.nstep != 1:
                          v.score += float(beam_logp[i, 0])
  
                      v.dec_out.append(beam_dec_out[i])
  
                      v.dec_state = self.decoder.select_state(
                          beam_state, i
                      )
  
                      if self.use_lm:
                          v.lm_state = beam_lm_states[i]
                          v.lm_scores = beam_lm_scores[i]
  
          kept_hyps = sorted(
              (S + V), 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 10 for calling the BeamSearchTransducerStreaming obj's underlying member method code with example usage: obj.nsc_beam_search(enc_out) and returns: self