code: def extract_cvss(
      data_folder,
      splits,
      kmeans_folder,
      encoder,
      layer,
      save_folder,
      sample_rate=16000,
      skip_extract=False,
  ):
      """
      Extract speech units for HiFi-GAN training on the CVSS datasets.
  
      Arguments
      ---------
      data_folder : str
          Path to the folder where the original CVSS dataset is stored.
      splits : list
          List of splits to prepare.
      kmeans_folder: str
          Path to the folder where the k-means model checkpoint is stored.
      encoder: str
          Url to the model used as feature extractor.
      layer: int
          Layer from which features are extracted.
      save_folder: str
          Path to the folder where the speech units are stored.
      sample_rate: int
          CVSS dataset sample rate
      skip_extract: Bool
          If True, skip extraction.
  
      Example
      -------
      >>> from recipes.CVSS.S2ST.extract_code import extract_cvss
      >>> data_folder = 'data/CVSS/'
      >>> splits = ['train', 'valid']
      >>> kmeans_folder = ./Quantization/results/kmeans/4321/save
      >>> encoder = facebook/hubert-base-ls960
      >>> layer = 6
      >>> save_folder = 'save/'
      >>> extract_cvss(data_folder, splits, kmeans_folder, encoder, layer, save_folder)
      """
      logger = setup_logger()
  
      if skip_extract:
          return
      # Create configuration for easily skipping code extraction stage
      conf = {
          "data_folder": data_folder,
          "splits": splits,
          "save_folder": save_folder,
          "kmeans_folder": kmeans_folder,
          "encoder": encoder,
          "layer": layer,
      }
  
      save_folder = pl.Path(save_folder)
      # Check if this phase is already done (if so, skip it)
      if skip(splits, save_folder, conf):
          logger.info(
              "Skipping code extraction, completed in previous run."
          )
          return
  
      # Fetch device
      device = get_device(use_cuda=True)
  
      save_opt = save_folder / OPT_FILE
      data_folder = pl.Path(data_folder)
  
      # Fetch K-means model
      kmeans_folder = pl.Path(kmeans_folder)
      kmeans_ckpt = kmeans_folder / "kmeans.ckpt"
      if not kmeans_ckpt.exists():
          logger.info(
              "K-means checkpoint not found, downloading it from HF."
          )
          kmeans_download_path = (
              save_folder / "pretrained_models/quantization"
          )
          kmeans_download_path.mkdir(exist_ok=True, parents=True)
          hf_hub_download(
              repo_id=kmeans_folder.as_posix(),
              filename="kmeans.ckpt",
              local_dir=kmeans_download_path,
          )
          kmeans_ckpt = kmeans_download_path / "kmeans.ckpt"
  
      encoder_save_path = save_folder / "pretrained_models"
      code_folder = save_folder / "codes"
      code_folder.mkdir(parents=True, exist_ok=True)
  
      logger.info(f"Loading encoder: {encoder} ...")
      encoder = Wav2Vec2(
          encoder,
          encoder_save_path.as_posix(),
          output_all_hiddens=True,
          output_norm=False,
          freeze_feature_extractor=True,
          freeze=True,
      ).to(device)
  
      # K-means model
      logger.info(f"Loading K-means model from {kmeans_ckpt} ...")
      kmeans_model = joblib.load(open(kmeans_ckpt, "rb"))
      kmeans_model.verbose = False
  
      for split in splits:
          dataset_path = data_folder / f"{split}.json"
          logger.info(f"Reading dataset from {dataset_path} ...")
          meta_json = json.load(open(dataset_path))
          for key in tqdm(meta_json.keys()):
              item = meta_json[key]
              wav = item["tgt_audio"]
              with torch.no_grad():
                  info = torchaudio.info(wav)
                  audio = sb.dataio.dataio.read_audio(wav)
                  audio = torchaudio.transforms.Resample(
                      info.sample_rate,
                      sample_rate,
                  )(audio)
                  audio = audio.unsqueeze(0).to(device)
                  feats = encoder.extract_features(audio)
                  feats = feats[layer]
                  feats = np_array(feats)
              pred = kmeans_model.predict(feats)
              np.save(code_folder / f"{key}_tgt.npy", pred)
  
      logger.info("Extraction completed.")
      save_pkl(conf, save_opt)
  
file_path: speechbrain/recipes/CVSS/S2ST/extract_code.py
imports: import logging
  import json
  import pathlib
  import joblib
  import torch
  import torchaudio
  import numpy
  from tqdm import tqdm
  import speechbrain
  from speechbrain.dataio.dataio import load_pkl, save_pkl
  from speechbrain.lobes.models.huggingface_transformers.wav2vec2 import Wav2Vec2
  from huggingface_hub import hf_hub_download
label_desc: Use image node_id 5 for calling a global function with example usage: extract_cvss(data_folder, splits, kmeans_folder, encoder, layer, save_folder, sample_rate, skip_extract) without return types