class_docstr: CTC Beam Search is a Beam Search for CTC which does not keep track of
  the blank and non-blank probabilities. Each new token probability is
  added to the general score, and each beams that share the same text are
  merged together.
  
  The implementation suppors n-gram scoring on words and SentencePiece tokens. The input
  is expected to be a log-probabilities tensor of shape [batch, time, vocab_size].
  
  The main advantage of this CTCBeamSearcher over the CTCPrefixBeamSearcher is that it is
  relatively faster, and obtains slightly better results. However, the implementation is
  based on the one from the PyCTCDecode toolkit, adpated for the SpeechBrain's needs and does
  not follow a specific paper. We do recommand to use the CTCPrefixBeamSearcher if you want
  to cite the appropriate paper for the decoding method.
  
  Several heuristics are implemented to speed up the decoding process:
  - pruning of the beam : the beams are pruned if their score is lower than
      the best beam score minus the beam_prune_logp
  - pruning of the tokens : the tokens are pruned if their score is lower than
      the token_prune_min_logp
  - pruning of the history : the beams are pruned if they are the same over
      max_ngram history
  - skipping of the blank : the frame is skipped if the blank probability is
      higher than the blank_skip_threshold
  
  Note: if the Acoustic Model is not trained, the Beam Search will
  take a lot of time. We do recommand to use Greedy Search during validation
  until the model is fully trained and ready to be evaluated on test sets.
  
  Arguments
  ---------
  **kwargs
      see CTCBaseSearcher, arguments are directly passed.
  
  Example
  -------
  >>> import torch
  >>> from speechbrain.decoders import CTCBeamSearcher
  >>> probs = torch.tensor([[[0.2, 0.0, 0.8],
  ...                   [0.4, 0.0, 0.6]]])
  >>> log_probs = torch.log(probs)
  >>> lens = torch.tensor([1.0])
  >>> blank_index = 2
  >>> vocab_list = ['a', 'b', '-']
  >>> searcher = CTCBeamSearcher(blank_index=blank_index, vocab_list=vocab_list)
  >>> hyps = searcher(probs, lens)
class_docstr_tok: ["CTC","Beam","Search","is","a","Beam","Search","for","CTC","which","does","not","keep","track","of","the","blank","and","non-blank","probabilities",".","Each","new","token","probability","is","added","to","the","general","score",",","and","each","beams","that","share","the","same","text","are","merged","together",".","The","implementation","suppors","n-gram","scoring","on","words","and","SentencePiece","tokens",".","The","input","is","expected","to","be","a","log-probabilities","tensor","of","shape","[","batch",",","time",",","vocab_size","]",".","The","main","advantage","of","this","CTCBeamSearcher","over","the","CTCPrefixBeamSearcher","is","that","it","is","relatively","faster",",","and","obtains","slightly","better","results",".","However",",","the","implementation","is","based","on","the","one","from","the","PyCTCDecode","toolkit",",","adpated","for","the","SpeechBrain","'s","needs","and","does","not","follow","a","specific","paper",".","We","do","recommand","to","use","the","CTCPrefixBeamSearcher","if","you","want","to","cite","the","appropriate","paper","for","the","decoding","method",".","Several","heuristics","are","implemented","to","speed","up","the","decoding","process",":","-","pruning","of","the","beam",":","the","beams","are","pruned","if","their","score","is","lower","than","the","best","beam","score","minus","the","beam_prune_logp","-","pruning","of","the","tokens",":","the","tokens","are","pruned","if","their","score","is","lower","than","the","token_prune_min_logp","-","pruning","of","the","history",":","the","beams","are","pruned","if","they","are","the","same","over","max_ngram","history","-","skipping","of","the","blank",":","the","frame","is","skipped","if","the","blank","probability","is","higher","than","the","blank_skip_threshold","Note",":","if","the","Acoustic","Model","is","not","trained",",","the","Beam","Search","will","take","a","lot","of","time",".","We","do","recommand","to","use","Greedy","Search","during","validation","until","the","model","is","fully","trained","and","ready","to","be","evaluated","on","test","sets",".","Arguments","--","--","--","--","-","*","*","kwargs","see","CTCBaseSearcher",",","arguments","are","directly","passed",".","Example","--","--","--","-",">",">",">","import","torch",">",">",">","from","speechbrain.decoders","import","CTCBeamSearcher",">",">",">","probs","=","torch.tensor","(","[","[","[","0.2",",","0.0",",","0.8","]",",","...","[","0.4",",","0.0",",","0.6","]","]","]",")",">",">",">","log_probs","=","torch.log","(","probs",")",">",">",">","lens","=","torch.tensor","(","[","1.0","]",")",">",">",">","blank_index","=","2",">",">",">","vocab_list","=","[","'","a","'",",","'","b","'",",","'-","'","]",">",">",">","searcher","=","CTCBeamSearcher","(","blank_index=blank_index",",","vocab_list=vocab_list",")",">",">",">","hyps","=","searcher","(","probs",",","lens",")"]
docstr: Perform CTC Prefix Beam Search decoding.
  
  If self.lm is not None, the language model scores are computed and added to the CTC scores.
  
  Arguments
  ---------
  log_probs : torch.Tensor
      The log probabilities of the CTC input.
      Shape: (seq_length, vocab_size)
  wav_len : int
      The length of the input sequence.
  beams : list
      The list of CTCBeam objects.
  cached_lm_scores : dict
      The cached language model scores.
  cached_p_lm_scores : dict
      The cached prefix language model scores.
  processed_frames : int
      The start frame of the current decoding step. (default: 0)
  
  Returns
  -------
  beams : list
      The list of CTCBeam objects.
docstr_tok: ["Perform","CTC","Prefix","Beam","Search","decoding",".","If","self.lm","is","not","None",",","the","language","model","scores","are","computed","and","added","to","the","CTC","scores",".","Arguments","--","--","--","--","-","log_probs",":","torch.Tensor","The","log","probabilities","of","the","CTC","input",".","Shape",":","(","seq_length",",","vocab_size",")","wav_len",":","int","The","length","of","the","input","sequence",".","beams",":","list","The","list","of","CTCBeam","objects",".","cached_lm_scores",":","dict","The","cached","language","model","scores",".","cached_p_lm_scores",":","dict","The","cached","prefix","language","model","scores",".","processed_frames",":","int","The","start","frame","of","the","current","decoding","step",".","(","default",":","0",")","Returns","--","--","--","-","beams",":","list","The","list","of","CTCBeam","objects","."]
code: def partial_decoding(
      self,
      log_probs: torch.Tensor,
      wav_len: int,
      beams: List[CTCBeam],
      cached_lm_scores: dict,
      cached_p_lm_scores: dict,
      processed_frames: int = 0,
  ) -> List[CTCBeam]:
      """Perform CTC Prefix Beam Search decoding.
  
      If self.lm is not None, the language model scores are computed and added to the CTC scores.
  
      Arguments
      ---------
      log_probs : torch.Tensor
          The log probabilities of the CTC input.
          Shape: (seq_length, vocab_size)
      wav_len : int
          The length of the input sequence.
      beams : list
          The list of CTCBeam objects.
      cached_lm_scores : dict
          The cached language model scores.
      cached_p_lm_scores : dict
          The cached prefix language model scores.
      processed_frames : int
          The start frame of the current decoding step. (default: 0)
  
      Returns
      -------
      beams : list
          The list of CTCBeam objects.
      """
      # select only the valid frames i.e. the frames that are not padded
      log_probs = log_probs[:wav_len]
  
      for frame_index, logit_col in enumerate(
          log_probs, start=processed_frames
      ):
          # skip the frame if the blank probability is higher than the threshold
          if logit_col[self.blank_index] > self.blank_skip_threshold:
              continue
  
          # get the tokens with the highest probability
          max_index = logit_col.argmax()
          tokens_index_list = set(
              np.where(logit_col > self.token_prune_min_logp)[0]
          ) | {max_index}
          new_beams = []
  
          # select tokens that are in the vocab
          # this is useful if the logit vocab_size is larger than the vocab_list
          tokens_index_list = tokens_index_list & set(
              range(len(self.vocab_list))
          )
  
          for token_index in tokens_index_list:
              p_token = logit_col[token_index]
              token = self.vocab_list[token_index]
  
              for beam in beams:
                  if (
                      token_index == self.blank_index
                      or beam.last_token == token
                  ):
                      if token_index == self.blank_index:
                          new_end_frame = beam.partial_frames[0]
                      else:
                          new_end_frame = frame_index + 1
  
                      new_part_frames = (
                          beam.partial_frames
                          if token_index == self.blank_index
                          else (beam.partial_frames[0], new_end_frame)
                      )
  
                      # if blank or repeated token, we only change the score
                      new_beams.append(
                          CTCBeam(
                              text=beam.text,
                              full_text=beam.full_text,
                              next_word=beam.next_word,
                              partial_word=beam.partial_word,
                              last_token=token,
                              last_token_index=token_index,
                              text_frames=beam.text_frames,
                              partial_frames=new_part_frames,
                              score=beam.score + p_token,
                          )
                      )
  
                  elif self.is_spm and token[:1] == self.spm_token:
                      # remove the spm token at the beginning of the token
                      clean_token = token[1:]
  
                      new_frame_list = (
                          beam.text_frames
                          if beam.partial_word == ""
                          else beam.text_frames + [beam.partial_frames]
                      )
  
                      # If the beginning of the token is the spm_token
                      # then it means that we are extending the beam with a new word.
                      # We need to change the new_word with the partial_word
                      # and reset the partial_word with the new token
                      new_beams.append(
                          CTCBeam(
                              text=beam.text,
                              full_text=beam.full_text,
                              next_word=beam.partial_word,
                              partial_word=clean_token,
                              last_token=token,
                              last_token_index=token_index,
                              text_frames=new_frame_list,
                              partial_frames=(
                                  frame_index,
                                  frame_index + 1,
                              ),
                              score=beam.score + p_token,
                          )
                      )
  
                  elif (
                      not self.is_spm
                      and token_index == self.space_index
                  ):
                      new_frame_list = (
                          beam.text_frames
                          if beam.partial_word == ""
                          else beam.text_frames + [beam.partial_frames]
                      )
  
                      # same as before but in the case of a non spm vocab
                      new_beams.append(
                          CTCBeam(
                              text=beam.text,
                              full_text=beam.full_text,
                              next_word=beam.partial_word,
                              partial_word="",
                              last_token=token,
                              last_token_index=token_index,
                              text_frames=new_frame_list,
                              partial_frames=(-1, -1),
                              score=beam.score + p_token,
                          )
                      )
                  else:
                      new_part_frames = (
                          (frame_index, frame_index + 1)
                          if beam.partial_frames[0] < 0
                          else (beam.partial_frames[0], frame_index + 1)
                      )
  
                      # last case, we are extending the partial_word with a new token
                      new_beams.append(
                          CTCBeam(
                              text=beam.text,
                              full_text=beam.full_text,
                              next_word=beam.next_word,
                              partial_word=beam.partial_word + token,
                              last_token=token,
                              last_token_index=token_index,
                              text_frames=beam.text_frames,
                              partial_frames=new_part_frames,
                              score=beam.score + p_token,
                          )
                      )
  
          # we merge the beams with the same text
          new_beams = self.merge_beams(new_beams)
  
          # kenlm scoring
          scored_beams = self.get_lm_beams(
              new_beams,
              cached_lm_scores,
              cached_p_lm_scores,
          )
  
          # remove beam outliers
          max_score = max([b.lm_score for b in scored_beams])
          scored_beams = [
              b
              for b in scored_beams
              if b.lm_score >= max_score + self.beam_prune_logp
          ]
  
          trimmed_beams = self.sort_beams(scored_beams)
  
          if self.prune_history:
              lm_order = 1 if self.lm is None else self.lm.order
              beams = self._prune_history(
                  trimmed_beams, lm_order=lm_order
              )
          else:
              beams = [CTCBeam.from_lm_beam(b) for b in trimmed_beams]
  
      return beams
  
imports: from itertools import groupby
  from speechbrain.dataio.dataio import length_to_mask
  import math
  import dataclasses
  import numpy
  import heapq
  import logging
  import torch
  import warnings
  from typing import Dict, List, Optional, Union, Any, Tuple
label_desc: Use image node_id 3 for calling the CTCBeamSearcher obj's underlying member method code with example usage: obj.partial_decoding(log_probs, wav_len, beams, cached_lm_scores, cached_p_lm_scores, processed_frames) and returns: beams