code: def test_beam_decoding_preserve_alignments(
      self, test_data_dir, beam_config
  ):
      beam_size = beam_config.pop("beam_size", 1)
      model, encoded, encoded_len = get_model_encoder_output(
          test_data_dir, "stt_en_conformer_transducer_small"
      )
      beam = beam_decode.BeamRNNTInfer(
          model.decoder,
          model.joint,
          beam_size=beam_size,
          return_best_hypothesis=False,
          preserve_alignments=True,
          **beam_config,
      )
  
      enc_out = encoded
      enc_len = encoded_len
      blank_id = torch.tensor(
          model.joint.num_classes_with_blank - 1, dtype=torch.int32
      )
  
      with torch.no_grad():
          hyps = beam(encoder_output=enc_out, encoded_lengths=enc_len)[
              0
          ]  # type: rnnt_utils.Hypothesis
          hyp, all_hyps = decode_text_from_nbest_hypotheses(
              hyps, model.decoding
          )
          hyp = hyp[0]  # best hypothesis
          all_hyps = all_hyps[0]
  
          assert hyp.alignments is not None
  
          if beam_config["search_type"] == "alsd":
              assert len(all_hyps) <= int(
                  beam_config["alsd_max_target_len"] * float(enc_len[0])
              )
  
          print("Beam search algorithm :", beam_config["search_type"])
          # Use the following commented print statements to check
          # the alignment of other algorithms compared to the default
          for idx, hyp_ in enumerate(
              all_hyps
          ):  # type: (int, rnnt_utils.Hypothesis)
              print("Hyp index", idx + 1, "text :", hyp_.text)
  
              # Alignment length (T) must match audio length (T)
              # NOTE: increase length threshold to two to prevent intermittent failures when a word is split into subwords
              assert abs(len(hyp_.alignments) - enc_len[0]) <= 2  # 1
  
              for t in range(len(hyp_.alignments)):
                  t_u = []
                  for u in range(len(hyp_.alignments[t])):
                      logp, label = hyp_.alignments[t][u]
                      assert torch.is_tensor(logp)
                      assert torch.is_tensor(label)
  
                      t_u.append(int(label))
  
                  # Blank token must be the last token in the current
                  if len(t_u) > 1:
                      assert t_u[-1] == blank_id
  
                      # No blank token should be present in the current timestep other than at the end
                      for token in t_u[:-1]:
                          assert token != blank_id
  
                  print(f"Tokens at timestep {t} = {t_u}")
              print()
  
              assert len(hyp_.timestep) > 0
              print("Timesteps", hyp_.timestep)
              print()
  
imports: import os
  from functools import lru_cache
  import pytest
  import torch
  from omegaconf import DictConfig
  from nemo.collections.asr.models import ASRModel
  from nemo.collections.asr.modules import RNNTDecoder, RNNTJoint
  from nemo.collections.asr.parts.mixins import mixins
  from nemo.collections.asr.parts.submodules import rnnt_beam_decoding
  from nemo.collections.asr.parts.submodules import rnnt_greedy_decoding
  from nemo.collections.asr.parts.submodules.rnnt_decoding import RNNTBPEDecoding, RNNTDecoding, RNNTDecodingConfig
  from nemo.collections.asr.parts.utils import rnnt_utils
  from nemo.core.utils import numba_utils
  from nemo.core.utils.numba_utils import __NUMBA_MINIMUM_VERSION__
label_desc: Use image node_id 5 for calling the TestRNNTDecoding obj's underlying member method code with example usage: obj.test_beam_decoding_preserve_alignments(test_data_dir, beam_config) without return types