class_docstr: Time synchronous beam search algorithm.
class_docstr_tok: ["Time","synchronous","beam","search","algorithm","."]
docstr: Execute a single time step.
docstr_tok: ["Execute","a","single","time","step","."]
code: def time_step(self, p_ctc: Any, ctc_score_dp: Any, hyps: Any) -> Any:
      """Execute a single time step."""
      pre_beam_threshold = np.sort(p_ctc)[-self.pre_beam_size]
      cands = set(np.where(p_ctc >= pre_beam_threshold)[0])
      if len(cands) == 0:
          cands = {np.argmax(p_ctc)}
      new_hyps = set()
      ctc_score_dp_next = defaultdict(
          lambda: (float("-inf"), float("-inf"))
      )  # (p_nb, p_b)
      tmp = []
      for hyp_l in hyps:
          p_prev_l = np.logaddexp(*ctc_score_dp[hyp_l])
          for c in cands:
              if c == self.blank:
                  logging.debug(
                      "blank cand, hypothesis is " + str(hyp_l)
                  )
                  p_nb, p_b = ctc_score_dp_next[hyp_l]
                  p_b = np.logaddexp(p_b, p_ctc[c] + p_prev_l)
                  ctc_score_dp_next[hyp_l] = (p_nb, p_b)
                  new_hyps.add(hyp_l)
              else:
                  l_plus = hyp_l + (int(c),)
                  logging.debug(
                      "non-blank cand, hypothesis is " + str(l_plus)
                  )
                  p_nb, p_b = ctc_score_dp_next[l_plus]
                  if c == hyp_l[-1]:
                      logging.debug(
                          "repeat cand, hypothesis is " + str(hyp_l)
                      )
                      p_nb_prev, p_b_prev = ctc_score_dp[hyp_l]
                      p_nb = np.logaddexp(p_nb, p_ctc[c] + p_b_prev)
                      p_nb_l, p_b_l = ctc_score_dp_next[hyp_l]
                      p_nb_l = np.logaddexp(
                          p_nb_l, p_ctc[c] + p_nb_prev
                      )
                      ctc_score_dp_next[hyp_l] = (p_nb_l, p_b_l)
                  else:
                      p_nb = np.logaddexp(p_nb, p_ctc[c] + p_prev_l)
                  if l_plus not in hyps and l_plus in ctc_score_dp:
                      p_b = np.logaddexp(
                          p_b,
                          p_ctc[self.blank]
                          + np.logaddexp(*ctc_score_dp[l_plus]),
                      )
                      p_nb = np.logaddexp(
                          p_nb, p_ctc[c] + ctc_score_dp[l_plus][0]
                      )
                      tmp.append(l_plus)
                  ctc_score_dp_next[l_plus] = (p_nb, p_b)
                  new_hyps.add(l_plus)
  
      scores = self.joint_score(new_hyps, ctc_score_dp_next)
  
      hyps = sorted(new_hyps, key=lambda ll: scores[ll], reverse=True)[
          : self.beam_size
      ]
      ctc_score_dp = ctc_score_dp_next.copy()
      return ctc_score_dp, hyps, scores
  
file_path: espnet/espnet/nets/beam_search_timesync.py
imports: import logging
  from collections import defaultdict
  from dataclasses import dataclass
  from typing import Any, Dict, List, Tuple
  import numpy
  import torch
  from espnet.nets.beam_search import Hypothesis
  from espnet.nets.scorer_interface import ScorerInterface
label_desc: Use image node_id 5 for calling the BeamSearchTimeSync obj's underlying member method code with example usage: obj.time_step(p_ctc, ctc_score_dp, hyps) and returns: ctc_score_dp,  hyps,  scores