code: def main(cfg: EvalBeamSearchNGramConfig):
      if is_dataclass(cfg):
          cfg = OmegaConf.structured(
              cfg
          )  # type: EvalBeamSearchNGramConfig
  
      valid_decoding_modes = [
          "greedy",
          "beamsearch",
          "beamsearch_ngram",
      ]
      if cfg.decoding_mode not in valid_decoding_modes:
          raise ValueError(
              f"Given decoding_mode={cfg.decoding_mode} is invalid. Available options are :\n"
              f"{valid_decoding_modes}"
          )
  
      if cfg.nemo_model_file.endswith(".nemo"):
          asr_model = nemo_asr.models.ASRModel.restore_from(
              cfg.nemo_model_file, map_location=torch.device(cfg.device)
          )
      else:
          logging.warning(
              "nemo_model_file does not end with .nemo, therefore trying to load a pretrained model with this name."
          )
          asr_model = nemo_asr.models.ASRModel.from_pretrained(
              cfg.nemo_model_file, map_location=torch.device(cfg.device)
          )
  
      target_transcripts = []
      manifest_dir = Path(cfg.input_manifest).parent
      with open(
          cfg.input_manifest, "r", encoding="utf_8"
      ) as manifest_file:
          audio_file_paths = []
          for line in tqdm(
              manifest_file,
              desc=f"Reading Manifest {cfg.input_manifest} ...",
              ncols=120,
          ):
              data = json.loads(line)
              audio_file = Path(data["audio_filepath"])
              if (
                  not audio_file.is_file()
                  and not audio_file.is_absolute()
              ):
                  audio_file = manifest_dir / audio_file
              target_transcripts.append(data["text"])
              audio_file_paths.append(str(audio_file.absolute()))
  
      punctuation_capitalization = PunctuationCapitalization(
          cfg.text_processing.punctuation_marks
      )
      if cfg.text_processing.do_lowercase:
          target_transcripts = punctuation_capitalization.do_lowercase(
              target_transcripts
          )
      if cfg.text_processing.rm_punctuation:
          target_transcripts = (
              punctuation_capitalization.rm_punctuation(
                  target_transcripts
              )
          )
      if cfg.text_processing.separate_punctuation:
          target_transcripts = (
              punctuation_capitalization.separate_punctuation(
                  target_transcripts
              )
          )
  
      if cfg.probs_cache_file and os.path.exists(cfg.probs_cache_file):
          logging.info(
              f"Found a pickle file of probabilities at '{cfg.probs_cache_file}'."
          )
          logging.info(
              f"Loading the cached pickle file of probabilities from '{cfg.probs_cache_file}' ..."
          )
          with open(cfg.probs_cache_file, "rb") as probs_file:
              all_probs = pickle.load(probs_file)
  
          if len(all_probs) != len(audio_file_paths):
              raise ValueError(
                  f"The number of samples in the probabilities file '{cfg.probs_cache_file}' does not "
                  f"match the manifest file. You may need to delete the probabilities cached file."
              )
      else:
  
          @contextlib.contextmanager
          def default_autocast():
              yield
  
          if cfg.use_amp:
              if (
                  torch.cuda.is_available()
                  and hasattr(torch.cuda, "amp")
                  and hasattr(torch.cuda.amp, "autocast")
              ):
                  logging.info("AMP is enabled!\n")
                  autocast = torch.cuda.amp.autocast
  
              else:
                  autocast = default_autocast
          else:
              autocast = default_autocast
  
          with autocast():
              with torch.no_grad():
                  if isinstance(asr_model, EncDecHybridRNNTCTCModel):
                      asr_model.cur_decoder = "ctc"
                  all_logits = asr_model.transcribe(
                      audio_file_paths,
                      batch_size=cfg.acoustic_batch_size,
                      logprobs=True,
                  )
  
          all_probs = all_logits
          if cfg.probs_cache_file:
              os.makedirs(
                  os.path.split(cfg.probs_cache_file)[0], exist_ok=True
              )
              logging.info(
                  f"Writing pickle files of probabilities at '{cfg.probs_cache_file}'..."
              )
              with open(cfg.probs_cache_file, "wb") as f_dump:
                  pickle.dump(all_probs, f_dump)
  
      wer_dist_greedy = 0
      cer_dist_greedy = 0
      words_count = 0
      chars_count = 0
      for batch_idx, probs in enumerate(all_probs):
          preds = np.argmax(probs, axis=1)
          preds_tensor = torch.tensor(preds, device="cpu").unsqueeze(0)
          if isinstance(asr_model, EncDecHybridRNNTCTCModel):
              pred_text = (
                  asr_model.ctc_decoding.ctc_decoder_predictions_tensor(
                      preds_tensor
                  )[0][0]
              )
          else:
              pred_text = asr_model._wer.decoding.ctc_decoder_predictions_tensor(
                  preds_tensor
              )[
                  0
              ][
                  0
              ]
  
          if cfg.text_processing.do_lowercase:
              pred_text = punctuation_capitalization.do_lowercase(
                  [pred_text]
              )[0]
          if cfg.text_processing.rm_punctuation:
              pred_text = punctuation_capitalization.rm_punctuation(
                  [pred_text]
              )[0]
          if cfg.text_processing.separate_punctuation:
              pred_text = (
                  punctuation_capitalization.separate_punctuation(
                      [pred_text]
                  )[0]
              )
  
          pred_split_w = pred_text.split()
          target_split_w = target_transcripts[batch_idx].split()
          pred_split_c = list(pred_text)
          target_split_c = list(target_transcripts[batch_idx])
  
          wer_dist = editdistance.eval(target_split_w, pred_split_w)
          cer_dist = editdistance.eval(target_split_c, pred_split_c)
  
          wer_dist_greedy += wer_dist
          cer_dist_greedy += cer_dist
          words_count += len(target_split_w)
          chars_count += len(target_split_c)
  
      logging.info(
          "Greedy WER/CER = {:.2%}/{:.2%}".format(
              wer_dist_greedy / words_count,
              cer_dist_greedy / chars_count,
          )
      )
  
      asr_model = asr_model.to("cpu")
  
      if cfg.decoding_mode == "beamsearch_ngram":
          if not os.path.exists(cfg.kenlm_model_file):
              raise FileNotFoundError(
                  f"Could not find the KenLM model file '{cfg.kenlm_model_file}'."
              )
          lm_path = cfg.kenlm_model_file
      else:
          lm_path = None
  
      # 'greedy' decoding_mode would skip the beam search decoding
      if cfg.decoding_mode in ["beamsearch_ngram", "beamsearch"]:
          if (
              cfg.beam_width is None
              or cfg.beam_alpha is None
              or cfg.beam_beta is None
          ):
              raise ValueError(
                  "beam_width, beam_alpha and beam_beta are needed to perform beam search decoding."
              )
          params = {
              "beam_width": cfg.beam_width,
              "beam_alpha": cfg.beam_alpha,
              "beam_beta": cfg.beam_beta,
          }
          hp_grid = ParameterGrid(params)
          hp_grid = list(hp_grid)
  
          best_wer_beam_size, best_cer_beam_size = None, None
          best_wer_alpha, best_cer_alpha = None, None
          best_wer_beta, best_cer_beta = None, None
          best_wer, best_cer = 1e6, 1e6
  
          logging.info(
              f"==============================Starting the beam search decoding==============================="
          )
          logging.info(f"Grid search size: {len(hp_grid)}")
          logging.info(f"It may take some time...")
          logging.info(
              f"=============================================================================================="
          )
  
          if cfg.preds_output_folder and not os.path.exists(
              cfg.preds_output_folder
          ):
              os.mkdir(cfg.preds_output_folder)
          for hp in hp_grid:
              if cfg.preds_output_folder:
                  preds_output_file = os.path.join(
                      cfg.preds_output_folder,
                      f"preds_out_width{hp['beam_width']}_alpha{hp['beam_alpha']}_beta{hp['beam_beta']}.tsv",
                  )
              else:
                  preds_output_file = None
  
              candidate_wer, candidate_cer = beam_search_eval(
                  asr_model,
                  cfg,
                  all_probs=all_probs,
                  target_transcripts=target_transcripts,
                  preds_output_file=preds_output_file,
                  lm_path=lm_path,
                  beam_width=hp["beam_width"],
                  beam_alpha=hp["beam_alpha"],
                  beam_beta=hp["beam_beta"],
                  beam_batch_size=cfg.beam_batch_size,
                  progress_bar=True,
                  punctuation_capitalization=punctuation_capitalization,
              )
  
              if candidate_cer < best_cer:
                  best_cer_beam_size = hp["beam_width"]
                  best_cer_alpha = hp["beam_alpha"]
                  best_cer_beta = hp["beam_beta"]
                  best_cer = candidate_cer
  
              if candidate_wer < best_wer:
                  best_wer_beam_size = hp["beam_width"]
                  best_wer_alpha = hp["beam_alpha"]
                  best_wer_beta = hp["beam_beta"]
                  best_wer = candidate_wer
  
          logging.info(
              f"Best WER Candidate = {best_wer:.2%} :: Beam size = {best_wer_beam_size}, "
              f"Beam alpha = {best_wer_alpha}, Beam beta = {best_wer_beta}"
          )
  
          logging.info(
              f"Best CER Candidate = {best_cer:.2%} :: Beam size = {best_cer_beam_size}, "
              f"Beam alpha = {best_cer_alpha}, Beam beta = {best_cer_beta}"
          )
          logging.info(
              f"================================================================================="
          )
  
imports: import contextlib
  import json
  import os
  import pickle
  from dataclasses import dataclass, field, is_dataclass
  from pathlib import Path
  from typing import List, Optional
  import editdistance
  import numpy
  import torch
  from omegaconf import MISSING, OmegaConf
  from sklearn.model_selection import ParameterGrid
  from tqdm.auto import tqdm
  import nemo.collections.asr
  from nemo.collections.asr.models import EncDecHybridRNNTCTCModel
  from nemo.collections.asr.parts.submodules import ctc_beam_decoding
  from nemo.collections.asr.parts.utils.transcribe_utils import PunctuationCapitalization, TextProcessingConfig
  from nemo.core.config import hydra_runner
  from nemo.utils import logging
label_desc: Use image node_id 2 for calling a global function with example usage: main(cfg) without return types