class_docstr: Beam search implementation for Transducer.
docstr: Time synchronous beam search implementation.
  
  Based on https://ieeexplore.ieee.org/document/9053040
  
  Args:
      enc_out: Encoder output sequence. (T, D)
  
  Returns:
      nbest_hyps: N-best hypothesis.
docstr_tok: ["Time","synchronous","beam","search","implementation",".","Based","on","https",":","\/\/ieeexplore.ieee.org\/document\/9053040","Args",":","enc_out",":","Encoder","output","sequence",".","(","T",",","D",")","Returns",":","nbest_hyps",":","N-best","hypothesis","."]
code: def time_sync_decoding(
      self, enc_out: torch.Tensor
  ) -> List[Hypothesis]:
      """Time synchronous beam search implementation.
  
      Based on https://ieeexplore.ieee.org/document/9053040
  
      Args:
          enc_out: Encoder output sequence. (T, D)
  
      Returns:
          nbest_hyps: N-best hypothesis.
  
      """
      beam = min(self.beam_size, self.vocab_size)
  
      beam_state = self.decoder.init_state(beam)
  
      B = [
          Hypothesis(
              yseq=[self.blank_id],
              score=0.0,
              dec_state=self.decoder.select_state(beam_state, 0),
          )
      ]
      cache = {}
  
      if self.use_lm:
          B[0].lm_state = self.lm.zero_state()
  
      for enc_out_t in enc_out:
          A = []
          C = B
  
          enc_out_t = enc_out_t.unsqueeze(0)
  
          for v in range(self.max_sym_exp):
              D = []
  
              (
                  beam_dec_out,
                  beam_state,
                  beam_lm_tokens,
              ) = self.decoder.batch_score(
                  C,
                  beam_state,
                  cache,
                  self.use_lm,
              )
  
              beam_logp = torch.log_softmax(
                  self.joint_network(enc_out_t, beam_dec_out),
                  dim=-1,
              )
              beam_topk = beam_logp[:, 1:].topk(beam, dim=-1)
  
              seq_A = [h.yseq for h in A]
  
              for i, hyp in enumerate(C):
                  if hyp.yseq not in seq_A:
                      A.append(
                          Hypothesis(
                              score=(
                                  hyp.score + float(beam_logp[i, 0])
                              ),
                              yseq=hyp.yseq[:],
                              dec_state=hyp.dec_state,
                              lm_state=hyp.lm_state,
                          )
                      )
                  else:
                      dict_pos = seq_A.index(hyp.yseq)
  
                      A[dict_pos].score = np.logaddexp(
                          A[dict_pos].score,
                          (hyp.score + float(beam_logp[i, 0])),
                      )
  
              if v < (self.max_sym_exp - 1):
                  if self.use_lm:
                      (
                          beam_lm_scores,
                          beam_lm_states,
                      ) = self.lm.batch_score(
                          beam_lm_tokens, [c.lm_state for c in C], None
                      )
  
                  for i, hyp in enumerate(C):
                      for logp, k in zip(
                          beam_topk[0][i], beam_topk[1][i] + 1
                      ):
                          new_hyp = Hypothesis(
                              score=(hyp.score + float(logp)),
                              yseq=(hyp.yseq + [int(k)]),
                              dec_state=self.decoder.select_state(
                                  beam_state, i
                              ),
                              lm_state=hyp.lm_state,
                          )
  
                          if self.use_lm:
                              new_hyp.score += (
                                  self.lm_weight * beam_lm_scores[i, k]
                              )
                              new_hyp.lm_state = beam_lm_states[i]
  
                          D.append(new_hyp)
  
              if self.score_norm_during:
                  C = sorted(
                      D,
                      key=lambda x: x.score / len(x.yseq),
                      reverse=True,
                  )[:beam]
              else:
                  C = sorted(D, key=lambda x: x.score, reverse=True)[
                      :beam
                  ]
  
          if self.score_norm_during:
              B = sorted(
                  A, key=lambda x: x.score / len(x.yseq), reverse=True
              )[:beam]
          else:
              B = sorted(A, key=lambda x: x.score, reverse=True)[:beam]
  
      return self.sort_nbest(B)
  
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 8 for calling the BeamSearchTransducerStreaming obj's underlying member method code with example usage: obj.time_sync_decoding(enc_out) and returns: self