code: def main(
      cfg: TranslationConfig,
  ) -> Union[TranslationConfig, List[str]]:
      logging.info(f"Hydra config: {OmegaConf.to_yaml(cfg)}")
  
      for key in cfg:
          cfg[key] = None if cfg[key] == "None" else cfg[key]
  
      if is_dataclass(cfg):
          cfg = OmegaConf.structured(cfg)
  
      if cfg.random_seed:
          pl.seed_everything(cfg.random_seed)
  
      if cfg.model_path is None and cfg.pretrained_name is None:
          raise ValueError(
              "Both cfg.model_path and cfg.pretrained_name cannot be None!"
          )
      if cfg.audio_dir is None and cfg.dataset_manifest is None:
          raise ValueError(
              "Both cfg.audio_dir and cfg.dataset_manifest cannot be None!"
          )
  
      # Load augmentor from exteranl yaml file which contains eval info, could be extend to other feature such VAD, P&C
      augmentor = None
      if cfg.eval_config_yaml:
          eval_config = OmegaConf.load(cfg.eval_config_yaml)
          augmentor = eval_config.test_ds.get("augmentor")
          logging.info(
              f"Will apply on-the-fly augmentation on samples during translation: {augmentor} "
          )
  
      # setup GPU
      if cfg.cuda is None:
          if torch.cuda.is_available():
              device = [0]  # use 0th CUDA device
              accelerator = "gpu"
              map_location = torch.device("cuda:0")
          elif (
              cfg.allow_mps
              and hasattr(torch.backends, "mps")
              and torch.backends.mps.is_available()
          ):
              logging.warning(
                  "MPS device (Apple Silicon M-series GPU) support is experimental."
                  " Env variable `PYTORCH_ENABLE_MPS_FALLBACK=1` should be set in most cases to avoid failures."
              )
              device = [0]
              accelerator = "mps"
              map_location = torch.device("mps")
          else:
              device = 1
              accelerator = "cpu"
              map_location = torch.device("cpu")
      else:
          device = [cfg.cuda]
          accelerator = "gpu"
          map_location = torch.device(f"cuda:{cfg.cuda}")
  
      logging.info(f"Inference will be done on device: {map_location}")
  
      asr_model, model_name = setup_model(cfg, map_location)
      trainer = pl.Trainer(devices=device, accelerator=accelerator)
      asr_model.set_trainer(trainer)
      asr_model = asr_model.eval()
  
      # collect additional translation information
      return_hypotheses = False
  
      # prepare audio filepaths and decide wether it's partial audio
      filepaths, partial_audio = prepare_audio_data(cfg)
  
      # setup AMP (optional)
      if (
          cfg.amp
          and torch.cuda.is_available()
          and hasattr(torch.cuda, "amp")
          and hasattr(torch.cuda.amp, "autocast")
      ):
          logging.info("AMP enabled!\n")
          autocast = torch.cuda.amp.autocast
      else:
  
          @contextlib.contextmanager
          def autocast():
              yield
  
      # Compute output filename
      cfg = compute_output_filename(cfg, model_name)
  
      # if translations should not be overwritten, and already exists, skip re-translation step and return
      if (
          not cfg.return_translations
          and not cfg.overwrite_translations
          and os.path.exists(cfg.output_filename)
      ):
          logging.info(
              f"Previous translations found at {cfg.output_filename}, and flag `overwrite_translations`"
              f"is {cfg.overwrite_translations}. Returning without re-translating text."
          )
          return cfg
  
      # translate audio
      with autocast():
          with torch.no_grad():
              translations = asr_model.translate(
                  paths2audio_files=filepaths,
                  batch_size=cfg.batch_size,
                  return_hypotheses=return_hypotheses,
              )
  
      logging.info(f"Finished translating {len(filepaths)} files !")
      logging.info(
          f"Writing translations into file: {cfg.output_filename}"
      )
  
      if cfg.return_translations:
          return translations
  
      # write audio translations
      with open(
          cfg.output_filename, "w", encoding="utf-8", newline="\n"
      ) as f:
          for filepath, translation in zip(filepaths, translations):
              item = {
                  "audio_filepath": filepath,
                  "pred_translation": translation,
              }
              f.write(json.dumps(item, ensure_ascii=False) + "\n")
      logging.info(
          f"Finished writing predictions to {cfg.output_filename}!"
      )
  
      return cfg
  
file_path: NeMo/examples/asr/speech_translation/translate_speech.py
imports: import contextlib
  import json
  import os
  from dataclasses import dataclass, is_dataclass
  from typing import List, Optional, Union
  import pytorch_lightning
  import torch
  from omegaconf import OmegaConf
  from nemo.collections.asr.modules.conformer_encoder import ConformerChangeConfig
  from nemo.collections.asr.parts.utils.transcribe_utils import compute_output_filename, prepare_audio_data, setup_model
  from nemo.core.config import hydra_runner
  from nemo.utils import logging
label_desc: Use image node_id 1 for calling a global function with example usage: main(cfg) and returns: cfg, cfg, translations