code: def main(cfg: BuildEnsembleConfig):
      # silencing all messages from nemo/ptl to avoid dumping tons of configs to the stdout
      logging.getLogger("pytorch_lightning").setLevel(logging.CRITICAL)
      logging.getLogger("nemo_logger").setLevel(logging.CRITICAL)
      LOG.info(f"Build ensemble config:\n{OmegaConf.to_yaml(cfg)}")
  
      # to ensure post init is called
      cfg = BuildEnsembleConfig(**cfg)
  
      pl.seed_everything(cfg.random_seed)
      cfg.transcription.random_seed = None  # seed is already applied
      cfg.transcription.return_transcriptions = True
      cfg.transcription.preserve_alignment = True
      cfg.transcription.ctc_decoding.temperature = cfg.temperature
      cfg.transcription.rnnt_decoding.temperature = cfg.temperature
      # this ensures that generated output is after log-softmax for consistency with CTC
  
      train_confidences = []
      dev_confidences = []
      train_labels = []
      dev_labels = []
  
      # registering clean-up function that will hold on to this list and
      # should clean up even if there is partial error in some of the transcribe
      # calls
      subsampled_manifests = []
      atexit.register(
          cleanup_subsampled_manifests, subsampled_manifests
      )
  
      # note that we loop over the same config.
      # This is intentional, as we need to run all models on all datasets
      # this loop will do the following things:
      # 1. Goes through each model X each training dataset
      # 2. Computes predictions by directly calling transcribe_speech.main
      # 3. Converts transcription to the confidence score(s) as specified in the config
      # 4. If dev sets are provided, computes the same for them
      # 5. Creates a list of ground-truth model indices by mapping each model
      #    to its own training dataset as specified in the config.
      # 6. After the loop, we either run tuning over all confidence scores or
      #    directly use a single score to fit logistic regression and save the
      #    final ensemble model.
      for model_idx, model_cfg in enumerate(cfg.ensemble):
          train_model_confidences = []
          dev_model_confidences = []
          for data_idx, data_cfg in enumerate(cfg.ensemble):
              if (
                  model_idx == 0
              ):  # generating subsampled manifests only one time
                  subsampled_manifests.append(
                      subsample_manifest(
                          data_cfg.training_manifest,
                          data_cfg.max_training_samples,
                      )
                  )
              subsampled_manifest = subsampled_manifests[data_idx]
  
              if model_cfg.model.endswith(".nemo"):
                  cfg.transcription.model_path = model_cfg.model
              else:  # assuming pretrained model
                  cfg.transcription.pretrained_name = model_cfg.model
  
              cfg.transcription.dataset_manifest = subsampled_manifest
  
              # training
              with tempfile.NamedTemporaryFile() as output_file:
                  cfg.transcription.output_filename = output_file.name
                  LOG.info(
                      "Transcribing training dataset %d with model %d",
                      data_idx,
                      model_idx,
                  )
                  transcriptions = transcribe_speech.main(
                      deepcopy(cfg.transcription)
                  )
                  LOG.info("Generating confidence scores")
                  # TODO: parallelize this loop?
                  for transcription in tqdm(transcriptions):
                      if cfg.tune_confidence:
                          train_model_confidences.append(
                              compute_all_confidences(
                                  transcription,
                                  cfg.tune_confidence_config,
                              )
                          )
                      else:
                          train_model_confidences.append(
                              compute_confidence(
                                  transcription, cfg.confidence
                              )
                          )
                      if (
                          model_idx == 0
                      ):  # labels are the same for all models
                          train_labels.append(data_idx)
  
              # optional dev
              if data_cfg.dev_manifest is not None:
                  cfg.transcription.dataset_manifest = (
                      data_cfg.dev_manifest
                  )
                  with tempfile.NamedTemporaryFile() as output_file:
                      cfg.transcription.output_filename = (
                          output_file.name
                      )
                      LOG.info(
                          "Transcribing dev dataset %d with model %d",
                          data_idx,
                          model_idx,
                      )
                      transcriptions = transcribe_speech.main(
                          deepcopy(cfg.transcription)
                      )
                      LOG.info("Generating confidence scores")
                      for transcription in tqdm(transcriptions):
                          if cfg.tune_confidence:
                              dev_model_confidences.append(
                                  compute_all_confidences(
                                      transcription,
                                      cfg.tune_confidence_config,
                                  )
                              )
                          else:
                              dev_model_confidences.append(
                                  compute_confidence(
                                      transcription, cfg.confidence
                                  )
                              )
                          if (
                              model_idx == 0
                          ):  # labels are the same for all models
                              dev_labels.append(data_idx)
  
          train_confidences.append(train_model_confidences)
          if dev_model_confidences:
              dev_confidences.append(dev_model_confidences)
  
      if cfg.tune_confidence:
          best_confidence, model_selection_block = find_best_confidence(
              train_confidences,
              train_labels,
              dev_confidences,
              dev_labels,
              cfg.tune_logistic_regression,
              cfg.tune_logistic_regression_config,
          )
      else:
          best_confidence = cfg.confidence
          # transposing with zip(*list)
          training_features = np.array(list(zip(*train_confidences)))
          training_labels = np.array(train_labels)
          if dev_confidences:
              dev_features = np.array(list(zip(*dev_confidences)))
              dev_labels = np.array(dev_labels)
          else:
              dev_features = None
              dev_labels = None
          model_selection_block, _ = train_model_selection(
              training_features,
              training_labels,
              dev_features,
              dev_labels,
              cfg.tune_logistic_regression,
              cfg.tune_logistic_regression_config,
              verbose=True,
          )
  
      with tempfile.TemporaryDirectory() as tmpdir:
          model_selection_block_path = os.path.join(
              tmpdir, "model_selection_block.pkl"
          )
          joblib.dump(model_selection_block, model_selection_block_path)
  
          # creating ensemble checkpoint
          ensemble_model = ConfidenceEnsembleModel(
              cfg=DictConfig(
                  {
                      "model_selection_block": model_selection_block_path,
                      "confidence": best_confidence,
                      "temperature": cfg.temperature,
                      "load_models": [
                          model_cfg.model for model_cfg in cfg.ensemble
                      ],
                  }
              ),
              trainer=None,
          )
          ensemble_model.save_to(cfg.output_path)
  
imports: import atexit
  import logging
  import os
  import random
  import sys
  import tempfile
  from copy import deepcopy
  from dataclasses import dataclass, field
  from pathlib import Path
  from typing import Dict, List, Optional, Tuple
  import joblib
  import numpy
  import pytorch_lightning
  from omegaconf import MISSING, DictConfig, OmegaConf
  from sklearn.linear_model import LogisticRegression
  from sklearn.metrics import confusion_matrix
  from sklearn.pipeline import Pipeline, make_pipeline
  from sklearn.preprocessing import StandardScaler
  from tqdm import tqdm
  from nemo.collections.asr.models.confidence_ensemble import ConfidenceEnsembleModel, ConfidenceSpec, compute_confidence, get_filtered_logprobs
  from nemo.collections.asr.parts.utils.asr_confidence_utils import ConfidenceConfig, ConfidenceMethodConfig, get_confidence_aggregation_bank, get_confidence_measure_bank
  from nemo.collections.asr.parts.utils.rnnt_utils import Hypothesis
  from nemo.core.config import hydra_runner
label_desc: Use image node_id 7 for calling a global function with example usage: main(cfg) without return types