code: def main(cfg: EvalBeamSearchNGramConfig):
      if is_dataclass(cfg):
          cfg = OmegaConf.structured(
              cfg
          )  # type: EvalBeamSearchNGramConfig
  
      valid_decoding_strategis = [
          "greedy_batch",
          "beam",
          "tsd",
          "alsd",
          "maes",
      ]
      if cfg.decoding_strategy not in valid_decoding_strategis:
          raise ValueError(
              f"Given decoding_strategy={cfg.decoding_strategy} is invalid. Available options are :\n"
              f"{valid_decoding_strategis}"
          )
  
      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)
          )
  
      if cfg.kenlm_model_file:
          if not os.path.exists(cfg.kenlm_model_file):
              raise FileNotFoundError(
                  f"Could not find the KenLM model file '{cfg.kenlm_model_file}'."
              )
          if cfg.decoding_strategy != "maes":
              raise ValueError(
                  f"Decoding with kenlm model is supported only for maes decoding algorithm."
              )
          lm_path = cfg.kenlm_model_file
      else:
          lm_path = None
          cfg.beam_alpha = [0.0]
      if cfg.hat_subtract_ilm:
          assert (
              lm_path
          ), "kenlm must be set for hat internal lm subtraction"
  
      if cfg.decoding_strategy != "maes":
          (
              cfg.maes_prefix_alpha,
              cfg.maes_expansion_gamma,
              cfg.hat_ilm_weight,
          ) = ([0], [0], [0])
  
      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()))
  
      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
  
          # manual calculation of encoder_embeddings
          with autocast():
              with torch.no_grad():
                  asr_model.eval()
                  asr_model.encoder.freeze()
                  device = next(asr_model.parameters()).device
                  all_probs = []
                  with tempfile.TemporaryDirectory() as tmpdir:
                      with open(
                          os.path.join(tmpdir, "manifest.json"),
                          "w",
                          encoding="utf-8",
                      ) as fp:
                          for audio_file in audio_file_paths:
                              entry = {
                                  "audio_filepath": audio_file,
                                  "duration": 100000,
                                  "text": "",
                              }
                              fp.write(json.dumps(entry) + "\n")
                      config = {
                          "paths2audio_files": audio_file_paths,
                          "batch_size": cfg.acoustic_batch_size,
                          "temp_dir": tmpdir,
                          "num_workers": cfg.num_workers,
                          "channel_selector": None,
                          "augmentor": None,
                      }
                      temporary_datalayer = (
                          asr_model._setup_transcribe_dataloader(config)
                      )
                      for test_batch in tqdm(
                          temporary_datalayer,
                          desc="Transcribing",
                          disable=True,
                      ):
                          encoded, encoded_len = asr_model.forward(
                              input_signal=test_batch[0].to(device),
                              input_signal_length=test_batch[1].to(
                                  device
                              ),
                          )
                          # dump encoder embeddings per file
                          for idx in range(encoded.shape[0]):
                              encoded_no_pad = encoded[
                                  idx, :, : encoded_len[idx]
                              ]
                              all_probs.append(encoded_no_pad)
  
          if cfg.probs_cache_file:
              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)
  
      if cfg.decoding_strategy == "greedy_batch":
          asr_model = asr_model.to("cpu")
          candidate_wer, candidate_cer = decoding_step(
              asr_model,
              cfg,
              all_probs=all_probs,
              target_transcripts=target_transcripts,
              beam_batch_size=cfg.beam_batch_size,
              progress_bar=True,
          )
          logging.info(
              f"Greedy batch WER/CER = {candidate_wer:.2%}/{candidate_cer:.2%}"
          )
  
      asr_model = asr_model.to("cpu")
  
      # 'greedy_batch' decoding_strategy would skip the beam search decoding
      if cfg.decoding_strategy in ["beam", "tsd", "alsd", "maes"]:
          if cfg.beam_width is None or cfg.beam_alpha is None:
              raise ValueError(
                  "beam_width and beam_alpha are needed to perform beam search decoding."
              )
          params = {
              "beam_width": cfg.beam_width,
              "beam_alpha": cfg.beam_alpha,
              "maes_prefix_alpha": cfg.maes_prefix_alpha,
              "maes_expansion_gamma": cfg.maes_expansion_gamma,
              "hat_ilm_weight": cfg.hat_ilm_weight,
          }
          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, best_cer = 1e6, 1e6
  
          logging.info(
              f"==============================Starting the {cfg.decoding_strategy} 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:
                  results_file = f"preds_out_{cfg.decoding_strategy}_bw{hp['beam_width']}"
                  if cfg.decoding_strategy == "maes":
                      results_file = f"{results_file}_ma{hp['maes_prefix_alpha']}_mg{hp['maes_expansion_gamma']}"
                      if cfg.kenlm_model_file:
                          results_file = (
                              f"{results_file}_ba{hp['beam_alpha']}"
                          )
                          if cfg.hat_subtract_ilm:
                              results_file = f"{results_file}_hat_ilmw{hp['hat_ilm_weight']}"
                  preds_output_file = os.path.join(
                      cfg.preds_output_folder, f"{results_file}.tsv"
                  )
              else:
                  preds_output_file = None
  
              cfg.decoding.beam_size = hp["beam_width"]
              cfg.decoding.ngram_lm_alpha = hp["beam_alpha"]
              cfg.decoding.maes_prefix_alpha = hp["maes_prefix_alpha"]
              cfg.decoding.maes_expansion_gamma = hp[
                  "maes_expansion_gamma"
              ]
              cfg.decoding.hat_ilm_weight = hp["hat_ilm_weight"]
  
              candidate_wer, candidate_cer = decoding_step(
                  asr_model,
                  cfg,
                  all_probs=all_probs,
                  target_transcripts=target_transcripts,
                  preds_output_file=preds_output_file,
                  beam_batch_size=cfg.beam_batch_size,
                  progress_bar=True,
              )
  
              if candidate_cer < best_cer:
                  best_cer_beam_size = hp["beam_width"]
                  best_cer_alpha = hp["beam_alpha"]
                  best_cer_ma = hp["maes_prefix_alpha"]
                  best_cer_mg = hp["maes_expansion_gamma"]
                  best_cer_hat_ilm_weight = hp["hat_ilm_weight"]
                  best_cer = candidate_cer
  
              if candidate_wer < best_wer:
                  best_wer_beam_size = hp["beam_width"]
                  best_wer_alpha = hp["beam_alpha"]
                  best_wer_ma = hp["maes_prefix_alpha"]
                  best_wer_ga = hp["maes_expansion_gamma"]
                  best_wer_hat_ilm_weight = hp["hat_ilm_weight"]
                  best_wer = candidate_wer
  
          wer_hat_parameter = ""
          if cfg.hat_subtract_ilm:
              wer_hat_parameter = (
                  f"HAT ilm weight = {best_wer_hat_ilm_weight}, "
              )
          logging.info(
              f"Best WER Candidate = {best_wer:.2%} :: Beam size = {best_wer_beam_size}, "
              f"Beam alpha = {best_wer_alpha}, {wer_hat_parameter}"
              f"maes_prefix_alpha = {best_wer_ma}, maes_expansion_gamma = {best_wer_ga} "
          )
  
          cer_hat_parameter = ""
          if cfg.hat_subtract_ilm:
              cer_hat_parameter = (
                  f"HAT ilm weight = {best_cer_hat_ilm_weight}"
              )
          logging.info(
              f"Best CER Candidate = {best_cer:.2%} :: Beam size = {best_cer_beam_size}, "
              f"Beam alpha = {best_cer_alpha}, {cer_hat_parameter} "
              f"maes_prefix_alpha = {best_cer_ma}, maes_expansion_gamma = {best_cer_mg}"
          )
          logging.info(
              f"================================================================================="
          )
  
imports: import contextlib
  import json
  import os
  import pickle
  import tempfile
  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.parts.submodules import rnnt_beam_decoding
  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