class_docstr: SSLDisentangler is a Conformer based model for extracting disentangled content and speaker embeddings
  from an audio waveform. This model uses a pre-trained Conformer SSL model. To extract the linguistic content 
  and speaker representations using a pre-trained Conformer, two randomly initialized downstream 
  heads are added and the entire setup is finetuned in multi-task manner for speech recognition and speaker verification. 
  These representations can be used by FastPitchModel_SSL for voice conversion by swapping the speaker embedding 
  of a given source utterance, with the speaker embedding of a target speaker.
class_docstr_tok: ["SSLDisentangler","is","a","Conformer","based","model","for","extracting","disentangled","content","and","speaker","embeddings","from","an","audio","waveform",".","This","model","uses","a","pre-trained","Conformer","SSL","model",".","To","extract","the","linguistic","content","and","speaker","representations","using","a","pre-trained","Conformer",",","two","randomly","initialized","downstream","heads","are","added","and","the","entire","setup","is","finetuned","in","multi-task","manner","for","speech","recognition","and","speaker","verification",".","These","representations","can","be","used","by","FastPitchModel_SSL","for","voice","conversion","by","swapping","the","speaker","embedding","of","a","given","source","utterance",",","with","the","speaker","embedding","of","a","target","speaker","."]
code: def training_step(self, batch, batch_idx):
      loss = 0.0
      optim_backbone, optim_downstream = self.optimizers()
      schedulers = self.lr_schedulers()
  
      for key in batch.keys():
          if key == "sv":
              signal = batch[key]["audio"]
              signal_len = batch[key]["audio_lens"]
              speaker_id = batch[key]["speaker_id"]
  
              sv_logits, sv_emb, _, _, _ = self.forward(
                  input_signal=signal, input_signal_length=signal_len
              )
              pred_speaker = torch.argmax(sv_logits, dim=1)
  
              sv_loss = self.sv_loss(
                  logits=sv_logits, labels=speaker_id
              )
              loss += sv_loss
              if not self._cfg.combined_loss:
                  optim_backbone.zero_grad()
                  optim_downstream.zero_grad()
                  self.manual_backward(sv_loss)
                  optim_backbone.step()
                  optim_downstream.step()
  
              correct = (
                  pred_speaker.eq(speaker_id.data.view_as(pred_speaker))
                  .sum()
                  .item()
              )
              acc = (correct / len(speaker_id)) * 100
  
              self.log("t_sv_loss", sv_loss.item())
              self.log("t_sv_accuracy", acc)
  
          elif key == "content":
              content_loss = 0
              signal = batch[key]["audio"]
              signal_len = batch[key]["audio_lens"]
              target = batch[key]["text"]  # (B, T)
              target_len = batch[key]["text_lens"]
  
              (
                  _,
                  _,
                  content_embedding,
                  content_log_probs,
                  encoded_len,
              ) = self.forward(
                  input_signal=signal, input_signal_length=signal_len
              )
  
              ctc_loss = self.ctc_loss(
                  content_log_probs, target, encoded_len, target_len
              )
              # check if ctc loss is nan
              if torch.isfinite(ctc_loss):
                  self.log("t_ctc_loss", ctc_loss.item())
                  content_loss += ctc_loss
              else:
                  logging.warning(f"ctc_loss is not finite")
  
              if self.pitch_augment:
                  augmented_signal = batch[key]["audio_shifted"]
                  if self.stop_gradient:
                      with torch.no_grad():
                          (
                              _,
                              _,
                              content_embedding_aug,
                              content_log_probs_aug,
                              _,
                          ) = self.forward(
                              input_signal=augmented_signal,
                              input_signal_length=signal_len,
                          )
                  else:
                      (
                          _,
                          _,
                          content_embedding_aug,
                          content_log_probs_aug,
                          _,
                      ) = self.forward(
                          input_signal=augmented_signal,
                          input_signal_length=signal_len,
                      )
                  if self.aug_loss_type == "mse":
                      sim_loss = self.mse_loss(
                          content_embedding, content_embedding_aug
                      )
                  elif self.aug_loss_type == "cosine":
                      cosine_similarity = (
                          torch.nn.functional.cosine_similarity(
                              content_embedding,
                              content_embedding_aug,
                              dim=-1,
                          ).mean()
                      )
  
                      sim_loss = 1.0 - cosine_similarity
  
                  content_loss += self._cfg.augment_sim_alpha * sim_loss
                  self.log("t_sim_loss", sim_loss.item())
  
                  if self.augment_ctc:
                      ctc_loss_aug = self.ctc_loss(
                          content_log_probs_aug,
                          target,
                          encoded_len,
                          target_len,
                      )
                      if torch.isfinite(ctc_loss_aug):
                          content_loss += ctc_loss_aug
                          self.log(
                              "t_ctc_loss_aug", ctc_loss_aug.item()
                          )
                      else:
                          logging.warning(
                              f"ctc_loss_aug is not finite. Add min duration to avoid getting here."
                          )
  
              loss += content_loss
  
              if not self._cfg.combined_loss:
                  optim_backbone.zero_grad()
                  optim_downstream.zero_grad()
                  self.manual_backward(content_loss)
                  optim_backbone.step()
                  optim_downstream.step()
  
              if isinstance(content_loss, torch.Tensor):
                  self.log("t_content_loss", content_loss.item())
  
      if self._cfg.combined_loss:
          optim_backbone.zero_grad()
          optim_downstream.zero_grad()
          self.manual_backward(loss)
          optim_backbone.step()
          optim_downstream.step()
  
      if schedulers is not None:
          sch1, sch2 = schedulers
          sch1.step()
          sch2.step()
  
      if self.trainer.global_step % 10 == 0:
          self.log("lr_backbone", optim_backbone.param_groups[0]["lr"])
          self.log(
              "lr_downstream", optim_downstream.param_groups[0]["lr"]
          )
          self.log("t_loss", loss)
  
imports: import itertools
  from typing import Iterable, Optional
  import editdistance
  import librosa
  import torch
  from hydra.utils import instantiate
  from omegaconf import DictConfig, OmegaConf
  from pytorch_lightning import Trainer
  from pytorch_lightning.loggers import TensorBoardLogger
  from pytorch_lightning.utilities.combined_loader import CombinedLoader
  from nemo.collections.asr.losses.angularloss import AngularSoftmaxLoss
  from nemo.collections.tts.data.dataset import TTSDataset
  from nemo.collections.tts.modules.ssl_tts import GreedyCTCDecoder
  from nemo.collections.tts.torch.tts_tokenizers import BaseTokenizer, EnglishCharsTokenizer
  from nemo.core.classes import ModelPT
  from nemo.core.classes.common import PretrainedModelInfo
  from nemo.core.optim.lr_scheduler import WarmupPolicy
  from nemo.utils import logging
  from nemo.utils.decorators import experimental
label_desc: Use image node_id 10 for calling the SSLDisentangler obj's underlying member method code with example usage: obj.training_step(batch, batch_idx) without return types