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 forward(
      self,
      input_signal=None,
      input_signal_length=None,
      normalize_content=True,
  ):
      (
          processed_signal,
          processed_signal_length,
      ) = self.preprocessor_disentangler(
          input_signal=input_signal,
          length=input_signal_length,
      )
  
      encoded, encoded_len = self.encoder(
          audio_signal=processed_signal, length=processed_signal_length
      )  # b,c,t
  
      for task in self._cfg.downstream_heads.task_names:
          if task == "speaker_verification":
              speaker_embedding = self.downstream_nets[
                  "speaker_verification"
              ](encoded[:, :, 0])
              l2_norm = torch.norm(
                  speaker_embedding, p=2, dim=-1, keepdim=True
              )
              speaker_embedding_normalized = speaker_embedding / l2_norm
              speaker_logits = self.sv_linear(
                  speaker_embedding_normalized
              )
  
          elif task == "content":
              encoded_btc = encoded.permute(0, 2, 1)
              content_embedding = self.downstream_nets["content"](
                  encoded_btc
              )
              if normalize_content:
                  l2_norm_content = torch.norm(
                      content_embedding, p=2, dim=-1, keepdim=True
                  )
                  content_embedding = (
                      content_embedding / l2_norm_content
                  )
  
              content_logits = self.content_linear(content_embedding)
              content_log_probs = content_logits.log_softmax(dim=2)
              content_log_probs = content_log_probs.permute(
                  1, 0, 2
              )  # t,b,c for ctc
  
          else:
              raise ValueError(
                  f"{task} is not a valid task. Task must be speaker_verification or content."
              )
  
      return (
          speaker_logits,
          speaker_embedding_normalized,
          content_embedding,
          content_log_probs,
          encoded_len,
      )
  
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 8 for calling the SSLDisentangler obj's underlying member method code with example usage: obj.forward(input_signal, input_signal_length, normalize_content) and returns: speaker_logits,  speaker_embedding_normalized,  content_embedding,  content_log_probs,  encoded_len