class_docstr: A class designed for performing ASR and diarization together.
  
  Attributes:
      cfg_diarizer (OmegaConf):
          Hydra config for diarizer key
      params (OmegaConf):
          Parameters config in diarizer.asr
      ctc_decoder_params (OmegaConf)
          Hydra config for beam search decoder
      realigning_lm_params (OmegaConf):
          Hydra config for realigning language model
      manifest_filepath (str):
          Path to the input manifest path
      nonspeech_threshold (float):
          Threshold for VAD logits that are used for creating speech segments
      fix_word_ts_with_VAD (bool):
          Choose whether to fix word timestamps by using VAD results
      root_path (str):
          Path to the folder where diarization results are saved
      vad_threshold_for_word_ts (float):
          Threshold used for compensating word timestamps with VAD output
      max_word_ts_length_in_sec (float):
          Maximum limit for the duration of each word timestamp
      word_ts_anchor_offset (float):
          Offset for word timestamps from ASR decoders
      run_ASR:
          Placeholder variable for an ASR launcher function
      realigning_lm:
          Placeholder variable for a loaded ARPA Language model
      ctm_exists (bool):
          Boolean that indicates whether all files have the corresponding reference CTM file
      frame_VAD (dict):
          Dictionary containing frame-level VAD logits
      AUDIO_RTTM_MAP:
          Dictionary containing the input manifest information
      color_palette (dict):
          Dictionary containing the ANSI color escape codes for each speaker label (speaker index)
class_docstr_tok: ["A","class","designed","for","performing","ASR","and","diarization","together",".","Attributes",":","cfg_diarizer","(","OmegaConf",")",":","Hydra","config","for","diarizer","key","params","(","OmegaConf",")",":","Parameters","config","in","diarizer.asr","ctc_decoder_params","(","OmegaConf",")","Hydra","config","for","beam","search","decoder","realigning_lm_params","(","OmegaConf",")",":","Hydra","config","for","realigning","language","model","manifest_filepath","(","str",")",":","Path","to","the","input","manifest","path","nonspeech_threshold","(","float",")",":","Threshold","for","VAD","logits","that","are","used","for","creating","speech","segments","fix_word_ts_with_VAD","(","bool",")",":","Choose","whether","to","fix","word","timestamps","by","using","VAD","results","root_path","(","str",")",":","Path","to","the","folder","where","diarization","results","are","saved","vad_threshold_for_word_ts","(","float",")",":","Threshold","used","for","compensating","word","timestamps","with","VAD","output","max_word_ts_length_in_sec","(","float",")",":","Maximum","limit","for","the","duration","of","each","word","timestamp","word_ts_anchor_offset","(","float",")",":","Offset","for","word","timestamps","from","ASR","decoders","run_ASR",":","Placeholder","variable","for","an","ASR","launcher","function","realigning_lm",":","Placeholder","variable","for","a","loaded","ARPA","Language","model","ctm_exists","(","bool",")",":","Boolean","that","indicates","whether","all","files","have","the","corresponding","reference","CTM","file","frame_VAD","(","dict",")",":","Dictionary","containing","frame-level","VAD","logits","AUDIO_RTTM_MAP",":","Dictionary","containing","the","input","manifest","information","color_palette","(","dict",")",":","Dictionary","containing","the","ANSI","color","escape","codes","for","each","speaker","label","(","speaker","index",")"]
docstr: Launch the diarization process using the given VAD timestamp (oracle_manifest).
  
  Args:
      diar_model_config (OmegaConf):
          Hydra configurations for speaker diarization
      word_and_timestamps (list):
          List containing words and word timestamps
  
  Returns:
      diar_hyp (dict):
          A dictionary containing rttm results which are indexed by a unique ID.
      score Tuple[pyannote object, dict]:
          A tuple containing pyannote metric instance and mapping dictionary between
          speakers in hypotheses and speakers in reference RTTM files.
docstr_tok: ["Launch","the","diarization","process","using","the","given","VAD","timestamp","(","oracle_manifest",")",".","Args",":","diar_model_config","(","OmegaConf",")",":","Hydra","configurations","for","speaker","diarization","word_and_timestamps","(","list",")",":","List","containing","words","and","word","timestamps","Returns",":","diar_hyp","(","dict",")",":","A","dictionary","containing","rttm","results","which","are","indexed","by","a","unique","ID",".","score","Tuple","[","pyannote","object",",","dict","]",":","A","tuple","containing","pyannote","metric","instance","and","mapping","dictionary","between","speakers","in","hypotheses","and","speakers","in","reference","RTTM","files","."]
code: def run_diarization(
      self, diar_model_config, word_timestamps
  ) -> Dict[str, List[str]]:
      """
      Launch the diarization process using the given VAD timestamp (oracle_manifest).
  
      Args:
          diar_model_config (OmegaConf):
              Hydra configurations for speaker diarization
          word_and_timestamps (list):
              List containing words and word timestamps
  
      Returns:
          diar_hyp (dict):
              A dictionary containing rttm results which are indexed by a unique ID.
          score Tuple[pyannote object, dict]:
              A tuple containing pyannote metric instance and mapping dictionary between
              speakers in hypotheses and speakers in reference RTTM files.
      """
  
      if diar_model_config.diarizer.asr.parameters.asr_based_vad:
          self._save_VAD_labels_list(word_timestamps)
          oracle_manifest = os.path.join(
              self.root_path, "asr_vad_manifest.json"
          )
          oracle_manifest = write_rttm2manifest(
              self.VAD_RTTM_MAP, oracle_manifest
          )
          diar_model_config.diarizer.vad.model_path = None
          diar_model_config.diarizer.vad.external_vad_manifest = (
              oracle_manifest
          )
  
      diar_model = ClusteringDiarizer(cfg=diar_model_config)
      score = diar_model.diarize()
      if (
          diar_model_config.diarizer.vad.model_path is not None
          and not diar_model_config.diarizer.oracle_vad
      ):
          self._get_frame_level_VAD(
              vad_processing_dir=diar_model.vad_pred_dir,
              smoothing_type=diar_model_config.diarizer.vad.parameters.smoothing,
          )
  
      diar_hyp = {}
      for k, audio_file_path in enumerate(self.audio_file_list):
          uniq_id = get_uniqname_from_filepath(audio_file_path)
          pred_rttm = os.path.join(
              self.root_path, "pred_rttms", uniq_id + ".rttm"
          )
          diar_hyp[uniq_id] = rttm_to_labels(pred_rttm)
      return diar_hyp, score
  
imports: import copy
  import csv
  import json
  import os
  from collections import OrderedDict
  from datetime import datetime
  from typing import Dict, List, Tuple
  import numpy
  from nemo.collections.asr.metrics.der import concat_perm_word_error_rate
  from nemo.collections.asr.metrics.wer import word_error_rate
  from nemo.collections.asr.models import ClusteringDiarizer
  from nemo.collections.asr.parts.utils.speaker_utils import audio_rttm_map, get_uniqname_from_filepath, labels_to_rttmfile, rttm_to_labels, write_rttm2manifest
  from nemo.utils import logging
label_desc: Use image node_id 10 for calling the OfflineDiarWithASR obj's underlying member method code with example usage: obj.run_diarization(diar_model_config, word_timestamps) and returns: diar_hyp,  score