Function preprocess

models/CLAP/training/data.py:566–649

Preprocess a single sample for wdsdataloader.

(
    sample,
    audio_ext,
    text_ext,
    max_len,
    audio_cfg,
    class_index_dict=None,
    data_filling="pad",
    data_truncating="rand_trunc",
    text_augment_selection=None,
)

def preprocess(
    sample,
    audio_ext,
    text_ext,
    max_len,
    audio_cfg,
    class_index_dict=None,
    data_filling="pad",
    data_truncating="rand_trunc",
    text_augment_selection=None,
):
    """
    Preprocess a single sample for wdsdataloader.
    """
    audio_data, orig_sr = sf.read(io.BytesIO(sample[audio_ext]))
    audio_data = int16_to_float32(float32_to_int16(audio_data))
    audio_data = torch.tensor(audio_data).float()

    # TODO: (yusong) to be include in the future
    # # if torchaudio not installed, use soundfile to load audio
    # if torchaudio is None:
    #     audio_data, orig_sr = sf.read(io.BytesIO(sample[audio_ext]))
    #     audio_data = torch.tensor(audio_data).float()
    # else:
    #     # https://github.com/webdataset/webdataset/blob/main/webdataset/autodecode.py
    #     with tempfile.TemporaryDirectory() as dirname:
    #         os.makedirs(dirname, exist_ok=True)
    #         fname = os.path.join(dirname, f"file.flac")
    #         with open(fname, "wb") as stream:
    #             stream.write(sample[audio_ext])
    #         audio_data, orig_sr = torchaudio.load(fname)
    #         audio_data = audio_data[0, :].float()

    sample = get_audio_features(
        sample, audio_data, max_len, data_truncating, data_filling, audio_cfg
    )
    del sample[audio_ext]

    try:
        json_dict_raw = json.loads(sample[text_ext].decode("utf-8"))
    except:
        print("sample[__url__]:", sample["__url__"])

    # For selecting augmented text from dataset
    if text_augment_selection is None or text_augment_selection == "none":
        texts = json_dict_raw["text"]
    elif text_augment_selection == "all":
        if "text_augment_all" in json_dict_raw.keys():
            texts = json_dict_raw["text_augment_all"]
        else:
            texts = json_dict_raw["text"]
    elif text_augment_selection == "augment_only":
        if "text_augment_all" in json_dict_raw.keys():
            if json_dict_raw["text_augment_t5"] is None:
                texts = json_dict_raw["text"]
            else:
                texts = json_dict_raw["text_augment_t5"]
        else: